springboot 整合 kafka 消息確認機制

在 Spring Boot 中整合 Kafka 消息確認機制，可以通過以下步驟實現：

1. 添加依賴

在 pom.xml 文件中添加 Kafka 和 Spring Boot Kafka 相關的依賴：

<dependencies>
    <!-- Spring Boot Kafka Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
</dependencies>

2. 配置 Kafka

在 application.yml 或 application.properties 文件中配置 Kafka 相關參數：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3. 創建消息生產者

創建一個消息生產者類，用于發送消息到 Kafka：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topic, String message) {
        kafkaTemplate.send(topic, message);
    }
}

4. 創建消息消費者

創建一個消息消費者類，用于從 Kafka 消費消息：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer {

    @Autowired
    private KafkaConsumer<String, String> kafkaConsumer;

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",
                record.key(), record.value(), record.partition(), record.offset());
    }
}

5. 啟用消息確認機制

在消費者類中啟用消息確認機制，可以通過實現 AcknowledgingMessageListener 接口來實現：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.AcknowledgingMessageListener;
import org.apache.kafka.clients.consumer.MessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaConsumer implements AcknowledgingMessageListener<String, String> {

    @Autowired
    private KafkaConsumer<String, String> kafkaConsumer;

    @Override
    public void onMessage(ConsumerRecord<String, String> record) {
        System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",
                record.key(), record.value(), record.partition(), record.offset());
        // 確認消息已處理
        getKafkaListener().acknowledge(record);
    }

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        onMessage(record);
    }

    @Override
    public void acknowledge(ConsumerRecord<String, String> record) {
        // 確認消息已處理
    }
}

通過以上步驟，你可以在 Spring Boot 中整合 Kafka 消息確認機制。當消費者接收到消息后，會調用 acknowledge 方法來確認消息已處理。