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 方法来确认消息已处理。