Kafka 核心知識

implementation 'org.springframework.kafka:spring-kafka'
// spring-boot-starter-parent 已管理版本，不需手動指定

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

spring:
  kafka:
    bootstrap-servers: localhost:9092   # Broker 地址，生產環境填多個

    producer:
      acks: all                          # 最安全：等待 ISR 全部確認
      retries: 3                         # 失敗自動重試
      enable-idempotence: true           # 冪等性：自動去重（Kafka 3.0+ 預設開啟）
      batch-size: 32768                  # 批次大小 32KB（預設 16KB）
      linger-ms: 5                       # 等待 5ms 累積批次（提升吞吐量）
      compression-type: lz4              # 壓縮格式（lz4 速度最快）
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

    consumer:
      group-id: order-service-group
      enable-auto-commit: false          # 必須手動提交！自動提交可能在業務失敗後仍 commit
      auto-offset-reset: latest          # latest=從最新訊息消費；earliest=從最舊訊息消費
      max-poll-records: 100              # 每次 poll 最多取 100 筆
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.example.dto"  # 信任反序列化的包路徑

    listener:
      ack-mode: manual_immediate         # 配合 enable-auto-commit: false

@Service
@RequiredArgsConstructor
public class OrderEventProducer {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void publishOrderCreated(Order order) {
        // 用 userId 作為 Key → 同一個用戶的訂單 Event 進入同一個 Partition
        // 保證該用戶的事件被同一個 Consumer 按序處理
        ListenableFuture<SendResult<String, Object>> future =
            kafkaTemplate.send("order-created", order.getUserId(), order);

        // 非同步 Callback：確認是否成功送達 Broker
        future.addCallback(
            result  -> log.info("訊息發送成功，Offset: {}", result.getRecordMetadata().offset()),
            failure -> log.error("訊息發送失敗: {}", failure.getMessage())
        );
    }

    // 事務性發送（需要 Exactly-Once）
    public void publishWithTransaction(Order order) {
        kafkaTemplate.executeInTransaction(t -> {
            t.send("order-created", order.getUserId(), order);
            t.send("audit-log",     order.getUserId(), new AuditLog(order));
            return true; // 兩條訊息要麼全成功，要麼全回滾
        });
    }
}

@Service
@RequiredArgsConstructor
@Slf4j
public class OrderEventConsumer {

    private final OrderService orderService;

    // concurrency = 3 → 啟動 3 個 Consumer 執行緒（不超過 Partition 數）
    @KafkaListener(
        topics   = "order-created",
        groupId  = "order-service-group",
        concurrency = "3"
    )
    public void handleOrderCreated(
            @Payload Order order,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        log.info("消費訊息 | partition={}, offset={}, orderId={}", partition, offset, order.getId());

        try {
            orderService.process(order); // 業務邏輯（必須具備冪等性！）
            ack.acknowledge();           // 業務成功才 commit Offset
        } catch (Exception e) {
            log.error("處理失敗，orderId={}, error={}", order.getId(), e.getMessage());
            // 不呼叫 ack.acknowledge() → Offset 不推進 → 下次仍會重試
            // 注意：重試次數達上限後應進入 DLQ，否則會永遠卡住
        }
    }
}

// 在 @Configuration 類別中配置
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory(
        ConsumerFactory<String, Object> consumerFactory,
        KafkaTemplate<String, Object> kafkaTemplate) {

    ConcurrentKafkaListenerContainerFactory<String, Object> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);

    // 設定重試策略：最多重試 3 次，每次間隔 2 秒
    factory.setCommonErrorHandler(
        new DefaultErrorHandler(
            // 達到最大重試次數後，將訊息轉發到 DLQ Topic
            new DeadLetterPublishingRecoverer(kafkaTemplate,
                (record, ex) -> new TopicPartition(record.topic() + ".DLQ", record.partition())
            ),
            new FixedBackOff(2000L, 3L) // 間隔 2s，最多重試 3 次
        )
    );

    return factory;
}

// 監聽 DLQ 的 Consumer（人工介入或告警）
@KafkaListener(topics = "order-created.DLQ", groupId = "dlq-group")
public void handleDLQ(ConsumerRecord<String, Object> record) {
    log.error("DLQ 收到問題訊息，Key={}, Value={}", record.key(), record.value());
    // 可以：發告警、儲存到 DB 供人工審查
}