• @KafkaListener注解详解(一)| 常用参数详解


    @KafkaListener 注解提供了许多可配置的参数,以便更灵活地定制 Kafka 消息监听器的行为。

    topics:

    描述: 指定监听的 Kafka 主题,可以是一个字符串数组。这是最基本的参数,它定义了监听器将从哪个或哪些主题接收消息。

    例子: @KafkaListener(topics = "my-topic")

    groupId:

    描述: 指定 Kafka 消费者组的 ID。每个消费者都有自己所属的组。一个组中可以有多个消费者。

    例子: @KafkaListener(groupId = "my-group", topics = "my-topic")

    id:

    描述: 每个Listener实例的重要标识。默认是一个自动生成的唯一 ID。如果不指定groupId,那么id将直接作为groupId。在同一应用中,如果有多个监听器,可以使用不同的id来标识不同的监听器容器。

    例子: @KafkaListener(id = "myListener", topics = "my-topic")

    concurrency:

    描述: 指定并发消费者的数量,即监听器容器的线程数。控制监听器的并发性,每个线程会创建一个消费者实例。较大的并发性可以提高消息处理的吞吐量。

    例子: @KafkaListener(topics = "my-topic", concurrency = "3")

    containerFactory:

    描述: 指定用于创建监听器容器的工厂类。可以用于配置监听器容器的属性。通过设置 containerFactory,可以更灵活地配置监听器容器的一些属性,例如消息转换器、错误处理器等。(移步另外一篇关于工厂类的文章

    例子: @KafkaListener(topics = "my-topic", containerFactory = "myContainerFactory")

    autoStartup:

    描述: 指定是否在启动时自动启动监听器容器。默认是 true。如果设置为false,则需要手动调用容器的start() 方法来启动监听器。(移步另外一篇关于改变kafka运行状态的文章

    例子: @KafkaListener(topics = "my-topic", autoStartup = "false")

    clientIdPrefix:

    描述: 指定 Kafka 消费者的客户端 ID 前缀。可以通过设置clientIdPrefix来自定义消费者的客户端 ID。

    例子: @KafkaListener(topics = "my-topic", clientIdPrefix = "my-client")

    containerGroup:

    描述: 指定监听器容器所属的组。如果有多个应用使用相同的消费者组,可以通过设置 containerGroup来区分它们。

    例子: @KafkaListener(topics = "my-topic", containerGroup = "my-group")

    errorHandler:

    描述: 指定错误处理器,用于处理监听器方法抛出的异常。定义一个错误处理器,可以在发生异常时进行自定义处理。

    例子: @KafkaListener(topics = "my-topic", errorHandler = "myErrorHandler")

    properties:

    描述: 指定其他的消费者配置属性,以键值对的形式提供。这种方式允许你通过注解的方式灵活地设置特定的消费者属性,而不必在全局配置文件中进行设置。请确保设置的属性是合法的 Kafka 消费者属性,并符合你的应用需求。

    例子: @KafkaListener(topics = "my-topic", properties = {"enable.auto.commit=false", "auto.offset.reset=earliest"})


    这些参数可以根据实际需求进行组合和配置,以满足特定场景的要求。例如,通过调整 concurrency 可以控制监听器的并发性,通过设置 autoStartup 可以控制监听器容器是否在应用启动时自动启动。其他参数也可以根据需要进行调整。

    DEMO

    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(topics = "my-topic", groupId = "my-group")
    7. public void listen(ConsumerRecord record) {
    8. System.out.println("Received message: " + record.value());
    9. }
    10. }

    关于ConsumerRecord record的介绍参考:url

    @KafkaListener注解应用于listen方法,指定要监听的Kafka主题和消费者组ID。

    在application.properties中配置Kafka连接信息:

    1. spring.kafka.bootstrap-servers=localhost:9092
    2. spring.kafka.consumer.group-id=my-group

    关于kafka在application.properties配置参数的介绍参考:url

  • 相关阅读:
    GAN Step By Step -- Step2 GAN的详细介绍及其应用
    【智能优化算法】基于象鼻虫优化算法求解单目标优化问题附matlab代码
    解决IDEA中maven无法引进okhttp依赖
    cad转pdf怎么转
    基于HTML+CSS+JavaScript制作一个介绍自己家乡河池主题的网站,排版整洁,内容丰富,主题鲜明
    【Make YOLO Great Again】YOLOv1-v7全系列大解析(Neck篇)
    学习平台助力职场发展与提升
    分析 NFT 项目的 5 个指标
    vscode检查更新菜单消失且不能自动更新(1.70.0不能自动更新到1.70.2)
    NVIDIA DPU — BlueField DPU Hardware
  • 原文地址:https://blog.csdn.net/maple05/article/details/134356357