在Spring Boot项目中使用@KafkaListener注解，可以方便地监听和处理从Kafka主题中流入的消息。以下是一个详细的实战指南：

引入依赖

首先，确保在Spring Boot项目中引入了Spring Kafka的相关依赖。在pom.xml中添加以下内容：

    
        org.springframework.kafka
        spring-kafka
        
        2.8.0
    
    
    
        org.springframework.kafka
        spring-kafka-test
        
        2.8.0
        
        test
    

配置Kafka连接

在application.yml或application.properties中配置Kafka连接信息：

spring:
  kafka:
    bootstrap-servers: localhost:9092 # Kafka集群地址
    consumer:
      group-id: my-consumer-group # 消费者组ID
      auto-offset-reset: earliest # 默认偏移量位置（earliest/latest）
      enable-auto-commit: false # 是否自动提交offset，一般建议关闭以便手动控制
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 键序列化器
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer # 值序列化器

创建消费者类

创建一个Java类，使用@KafkaListener注解标注监听方法，该方法将在监听到相应主题的消息时被调用。

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaConsumer {
    @KafkaListener(topics = "my-topic")
    public void listenToMessage(String message) {
        System.out.println("Received message from Kafka: " + message);
        // 在这里处理消息逻辑
    }
    // 如果需要监听多个主题，可以添加更多的@KafkaListener注解方法
    @KafkaListener(topics = "#{'${kafka.topics}'.split(',')}")
    public void listenToMultipleTopics(String message) {
        // ...
    }
}

在上面的示例中：

• @Component注解表明这是一个Spring Bean，因此会被Spring容器管理。
• @KafkaListener(topics = "my-topic")注解标记了监听的主题名为my-topic的方法。
• 方法参数类型根据实际消息内容进行设定，这里假设是字符串类型。
• 若需要动态监听多个主题，可以使用SpEL表达式从配置文件中读取主题列表。

  动态主题和并发消费

  • 动态主题：可以通过SpEL表达式或者其他Spring Expression Language的方式来动态指定监听的主题。

  • 并发消费：默认情况下，每个@KafkaListener注解方法将会在一个单独的线程中运行。如果希望在同一方法中并发处理多个主题或者分区，可以配置concurrency属性：

    @KafkaListener(id = "multi-topic-listener", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "3")
    public void listenToMultipleTopicsConcurrently(String message) {
        // ...
    }
    

    这里的concurrency = "3"意味着会创建3个工作线程并发处理来自指定主题的消息。

    手动提交偏移量

    如果在配置中关闭了自动提交，那么在处理完消息后需要手动提交偏移量：

    import org.springframework.kafka.support.Acknowledgment;
    import org.springframework.kafka.annotation.KafkaListener;
    @KafkaListener(topics = "my-topic")
    public void listenToMessage(String message, Acknowledgment acknowledgment) {
        System.out.println("Processing message: " + message);
        // ... 处理消息逻辑
        acknowledgment.acknowledge(); // 手动提交偏移量
    }
    

    错误处理与重试

    可以配置监听方法的异常处理器，以及消息重试策略，这些都是Spring Kafka提供的高级功能，可以根据业务需求进行配置。