当前位置:   article > 正文

Spring Kafka 之 @KafkaListener 注解详解_@kafkalistener groupid

@kafkalistener groupid

       我们在开发的过程中当使用到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")
  • 说明:如果不指定 groupIdid 将直接作为 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 连接的基本信息。

三、示例

  1. import org.apache.kafka.clients.consumer.ConsumerRecord;
  2. import org.springframework.kafka.annotation.KafkaListener;
  3. import org.springframework.stereotype.Service;
  4. @Service
  5. public class KafkaConsumerService {
  6. // 使用 @KafkaListener 注解配置 Kafka 消息监听器
  7. @KafkaListener(
  8. topics = "my-topic", // 监听名为 "my-topic" 的 Kafka 主题
  9. groupId = "my-consumer-group", // 消费者组 ID 为 "my-consumer-group"
  10. id = "myListener", // 监听器实例的唯一标识符为 "myListener"
  11. containerFactory = "kafkaListenerContainerFactory", // 使用名为 "kafkaListenerContainerFactory" 的工厂 bean 来创建 MessageListenerContainer
  12. autoStartup = "true" // 应用程序启动时自动启动监听器,默认为 true,这里显式指定
  13. )
  14. public void consumeMessage(ConsumerRecord<?, ?> record) {
  15. // 处理接收到的 Kafka 消息
  16. String topic = record.topic();
  17. String value = (String) record.value();
  18. System.out.println("Received message from topic: " + topic + ", value: " + value);
  19. }
  20. // (可选)你可以通过配置类来定义 kafkaListenerContainerFactory
  21. // 通常在 @Configuration 类中定义 Bean
  22. // ...
  23. // @Bean
  24. // public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
  25. // ConsumerFactory<?, ?> consumerFactory) {
  26. // // 自定义 Kafka 监听器容器工厂
  27. // ...
  28. // }
  29. // 注意:上面的配置类代码是示例性的,并未完整展示如何配置一个 KafkaListenerContainerFactory。
  30. // 实际的配置将依赖于你的应用程序和 Kafka 配置需求。
  31. }

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/花生_TL007/article/detail/728980
推荐阅读
相关标签
  

闽ICP备14008679号