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")
  • 说明:如果不指定 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 连接的基本信息。

三、示例

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 配置需求。  
}

相关推荐

  1. Spring Kafka @KafkaListener 注解详解

    2024-06-06 11:00:03       30 阅读
  2. Kafka的@KafkaListener注解参数详解

    2024-06-06 11:00:03       51 阅读
  3. springboot2.2.9整合kafkaKafkaListener实现原理

    2024-06-06 11:00:03       42 阅读

最近更新

  1. docker php8.1+nginx base 镜像 dockerfile 配置

    2024-06-06 11:00:03       98 阅读
  2. Could not load dynamic library ‘cudart64_100.dll‘

    2024-06-06 11:00:03       106 阅读
  3. 在Django里面运行非项目文件

    2024-06-06 11:00:03       87 阅读
  4. Python语言-面向对象

    2024-06-06 11:00:03       96 阅读

热门阅读

  1. 泛微OA调用发送消息接口 .Net C#示例

    2024-06-06 11:00:03       25 阅读
  2. .NET Redis限制接口请求频率 滑动窗口算法

    2024-06-06 11:00:03       27 阅读
  3. electron-vite打包成安装包配置

    2024-06-06 11:00:03       31 阅读
  4. 爬山算法的详细介绍

    2024-06-06 11:00:03       34 阅读
  5. Android12.0 SIM卡语言自适应

    2024-06-06 11:00:03       20 阅读
  6. web 预览显示本地图片、音频

    2024-06-06 11:00:03       34 阅读
  7. 简单几步,用Python实现VPN搭建

    2024-06-06 11:00:03       39 阅读
  8. 数据仓库分层模型

    2024-06-06 11:00:03       32 阅读
  9. Hyperf 框架常见面试题

    2024-06-06 11:00:03       32 阅读