Spring Boot整合Kafka实现消息消费的完整示例

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

spring:
  kafka:
    bootstrap-servers: localhost:9092 # 你的Kafka服务器地址
    consumer:
      group-id: my-spring-consumer-group # 消费者组ID，非常重要
      auto-offset-reset: latest # 当没有初始偏移量或当前偏移量无效时，从哪里开始消费：earliest/latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      enable-auto-commit: false # 推荐手动提交，以保证消息处理的可靠性
      max-poll-records: 500 # 每次poll操作最多获取的记录数
      fetch-min-bytes: 1 # 消费者从服务器获取数据时，最小等待字节数
      fetch-max-wait: 500 # 消费者从服务器获取数据时，最大等待时间（毫秒）

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MyKafkaConsumer {

    // 监听名为 "my-topic" 的主题
    @KafkaListener(topics = "my-topic", groupId = "my-spring-consumer-group", containerFactory = "kafkaListenerContainerFactory")
    public void listen(String message) {
        System.out.println("收到消息 (字符串): " + message);
        // 如果是自动提交，这里就结束了
    }

    // 监听并手动提交偏移量，这在生产环境非常常见
    @KafkaListener(topics = "my-topic-manual-ack", groupId = "my-spring-consumer-group", containerFactory = "kafkaListenerContainerFactory")
    public void listenWithManualAck(ConsumerRecord<String, String> record, Acknowledgment ack) {
        System.out.println("收到消息 (手动ack): Key=" + record.key() + ", Value=" + record.value() + ", Offset=" + record.offset());
        try {
            // 模拟业务处理
            Thread.sleep(100);
            if (record.value().contains("error")) {
                throw new RuntimeException("模拟处理错误");
            }
            // 业务处理成功后，手动提交偏移量
            ack.acknowledge();
            System.out.println("消息处理成功并提交偏移量: " + record.offset());
        } catch (Exception e) {
            System.err.println("消息处理失败，不提交偏移量: " + record.offset() + " 错误: " + e.getMessage());
            // 实际生产中，这里可能需要记录日志，或者将消息发送到死信队列 (DLQ)
            // 不调用 ack.acknowledge()，下次Kafka会重新投递这条消息
        }
    }

    // 批量消费，可以显著提高吞吐量
    @KafkaListener(topics = "my-topic-batch", groupId = "my-spring-consumer-group", containerFactory = "batchKafkaListenerContainerFactory")
    public void listenBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        System.out.println("收到批量消息，共 " + records.size() + " 条");
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("  - Key=" + record.key() + ", Value=" + record.value() + ", Offset=" + record.offset());
            // 批量处理逻辑
        }
        // 批量处理完成后，一次性提交整个批次的偏移量
        ack.acknowledge();
        System.out.println("批量消息处理完成并提交偏移量");
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.beans.factory.annotation.Value;
import java.util.HashMap;
import java.util.Map;

@EnableKafka // 启用Kafka功能
@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Value("${spring.kafka.consumer.auto-offset-reset}")
    private String autoOffsetReset;

    @Value("${spring.kafka.consumer.key-deserializer}")
    private String keyDeserializer;

    @Value("${spring.kafka.consumer.value-deserializer}")
    private String valueDeserializer;

    @Value("${spring.kafka.consumer.enable-auto-commit}")
    private Boolean enableAutoCommit;

    // 这是一个通用的消费者工厂配置，用于手动提交偏移量
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
        props.put(org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
        // 如果要处理JSON，这里可以配置JsonDeserializer
        // props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); // 允许反序列化所有包下的类，生产环境慎用
        return new DefaultKafkaConsumerFactory<>(props);
    }

    // 普通的Kafka监听器容器工厂，用于单条消息消费
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 设置为手动提交偏移量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        // 设置并发消费者数量，根据实际情况调整
        factory.setConcurrency(3); // 比如有3个分区，可以设置3个并发消费者
        return factory;
    }

    // 批量消费的Kafka监听器容器工厂
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true); // 启用批量监听
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.setConcurrency(2);
        return factory;
    }
}