Spring Kafka 之 @KafkaListener 注解详解
我们在开发的过程中当使用到kafka监听消费的时候会使用到@KafkaListener注解，下面我们就介绍下它的常见属性和使用。
一、介绍
@KafkaListener 是 Spring Kafka 提供的一个注解，用于声明一个方法作为 Kafka 消息的监听器
二、主要参数
1、topic
- 描述：指定监听的 Kafka 主题，可以是一个字符串数组。
- 示例：
@KafkaListener(topics = "my-topic") - 说明：定义了监听器将从哪个或哪些主题接收消息。
2、groupId
- 描述：指定 Kafka 消费者组的 ID。
- 示例：
@KafkaListener(groupId = "my-group", topics = "my-topic") - 说明：每个消费者都有自己所属的组。一个组中可以有多个消费者，它们共同处理消息。
3、id
- 描述：每个 Listener 实例的唯一标识符。
- 示例：
@KafkaListener(id = "myListener", topics = "my-topic") - 说明：如果不指定
groupId，id将直接作为groupId。在多监听器的应用中，可以使用不同的id来区分不同的监听器容器。
4、containerFactory
- 描述：指定用于创建
MessageListenerContainer的工厂 bean 的名称。 - 示例：
@KafkaListener(containerFactory = "yourContainerFactory", topics = "your-topic") - 说明：容器负责管理消息监听器的生命周期和线程管理。
5、topicPattern
- 描述：指定一个正则表达式模式，用于匹配要监听的多个主题。
- 示例：
@KafkaListener(topicPattern = "your-topic.*", groupId = "your-group-id") - 说明：允许通过模式来匹配一组相关的主题。
6、autoStartup
- 描述：指定是否在应用程序启动时自动启动监听器。
- 示例：
@KafkaListener(autoStartup = "false", topics = "your-topic") - 说明：默认为
true，可以手动控制监听器的启动和停止。
7、bootstrap.servers（注意：这不是 @KafkaListener 的直接参数，但通常在 Kafka 配置中指定）
- 描述：Kafka 服务器的地址列表，用于连接到 Kafka 集群。
- 示例：
"localhost:9092,anotherhost:9092" - 说明：用于配置 Kafka 连接的基本信息。
三、示例
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service; @Service
public class KafkaConsumerService { // 使用 @KafkaListener 注解配置 Kafka 消息监听器 @KafkaListener( topics = "my-topic", // 监听名为 "my-topic" 的 Kafka 主题 groupId = "my-consumer-group", // 消费者组 ID 为 "my-consumer-group" id = "myListener", // 监听器实例的唯一标识符为 "myListener" containerFactory = "kafkaListenerContainerFactory", // 使用名为 "kafkaListenerContainerFactory" 的工厂 bean 来创建 MessageListenerContainer autoStartup = "true" // 应用程序启动时自动启动监听器，默认为 true，这里显式指定 ) public void consumeMessage(ConsumerRecord<?, ?> record) { // 处理接收到的 Kafka 消息 String topic = record.topic(); String value = (String) record.value(); System.out.println("Received message from topic: " + topic + ", value: " + value); } // （可选）你可以通过配置类来定义 kafkaListenerContainerFactory // 通常在 @Configuration 类中定义 Bean // ... // @Bean // public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory( // ConsumerFactory<?, ?> consumerFactory) { // // 自定义 Kafka 监听器容器工厂 // ... // } // 注意：上面的配置类代码是示例性的，并未完整展示如何配置一个 KafkaListenerContainerFactory。 // 实际的配置将依赖于你的应用程序和 Kafka 配置需求。
}