【spring-kafka】@KafkaListener 详解与使用

2022 年 9 月 15 日
江西
本文字数：3566 字
阅读完需：约 12 分钟

作者：石臻臻,CSDN 博客之星 Top5、Kafka Contributor、nacos Contributor、华为云 MVP,腾讯云 TVP,滴滴 Kafka 技术专家、 KnowStreaming。
KnowStreaming 是滴滴开源的Kafka运维管控平台, 有兴趣一起参与参与开发的同学,但是怕自己能力不够的同学,可以联系我,当你导师带你参与开源！ 。

1 说明

从 2.2.4 版开始，您可以直接在注释上指定 Kafka 使用者属性，这些属性将覆盖在使用者工厂中配置的具有相同名称的所有属性。您不能通过这种方式指定 group.id 和 client.id 属性。他们将被忽略；
可以使用 #{…}或属性占位符（${…}）在 SpEL 上配置注释上的大多数属性。比如:

   @KafkaListener(id = "consumer-id",topics = "SHI_TOPIC1",concurrency = "${listen.concurrency:3}",            clientIdPrefix = "myClientId")

复制代码

属性concurrency将会从容器中获取listen.concurrency的值,如果不存在就默认用 3

2@KafkaListener 详解

id 监听器的 id

①. 消费者线程命名规则

填写:

2020-11-19 14:24:15 c.d.b.k.KafkaListeners 120 [INFO] 线程:Thread[consumer-id5-1-C-1,5,main]-groupId:BASE-DEMO consumer-id5 消费

没有填写 ID:

2020-11-19 10:41:26 c.d.b.k.KafkaListeners 137 [INFO] 线程:Thread[org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1,5,main] consumer-id7

②.在相同容器中的监听器 ID 不能重复

否则会报错

Caused by: java.lang.IllegalStateException: Another endpoint is already registered with id

复制代码

③.会覆盖消费者工厂的消费组 GroupId

假如配置文件属性配置了消费组kafka.consumer.group-id=BASE-DEMO正常情况它是该容器中的默认消费组但是如果设置了 @KafkaListener(id = "consumer-id7", topics = {"SHI_TOPIC3"})那么当前消费者的消费组就是consumer-id7 ;

当然如果你不想要他作为 groupId 的话可以设置属性idIsGroup = false;那么还是会使用默认的 GroupId;

④. 如果配置了属性 groupId,则其优先级最高

 @KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3",groupId = "groupId-test")

复制代码

例如上面代码中最终这个消费者的消费组GroupId是 "groupId-test"

该 id 属性（如果存在）将用作 Kafka 消费者 group.id 属性,并覆盖消费者工厂中的已配置属性（如果存在）您还可以 groupId 显式设置或将其设置 idIsGroup 为 false，以恢复使用使用者工厂的先前行为 group.id。

groupId 消费组名

指定该消费组的消费组名; 关于消费组名的配置可以看看上面的 id 监听器的 id

如何获取消费者 group.id

在监听器中调用KafkaUtils.getConsumerGroupId()可以获得当前的 groupId; 可以在日志中打印出来; 可以知道是哪个客户端消费的;

topics 指定要监听哪些 topic(与 topicPattern、topicPartitions 三选一)

可以同时监听多个topics = {"SHI_TOPIC3","SHI_TOPIC4"}

topicPattern 匹配 Topic 进行监听(与 topics、topicPartitions 三选一)

topicPartitions 显式分区分配

可以为监听器配置明确的主题和分区（以及可选的初始偏移量）

@KafkaListener(id = "thing2", topicPartitions =        { @TopicPartition(topic = "topic1", partitions = { "0", "1" }),          @TopicPartition(topic = "topic2", partitions = "0",             partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))        })public void listen(ConsumerRecord<?, ?> record) {    ...}

复制代码

上面例子意思是监听topic1的 0,1 分区；监听topic2的第 0 分区,并且第 1 分区从 offset 为 100 的开始消费;

errorHandler 异常处理

实现KafkaListenerErrorHandler; 然后做一些异常处理;

@Componentpublic class KafkaDefaultListenerErrorHandler implements KafkaListenerErrorHandler {    @Override    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {        return null;    }
    @Override    public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {     //do someting        return null;    }}

复制代码

调用的时候填写 beanName;例如errorHandler="kafkaDefaultListenerErrorHandler"

containerFactory 监听器工厂

指定生成监听器的工厂类;

例如我写一个批量消费的工厂类

    /**     * 监听器工厂 批量消费     * @return     */    @Bean    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> batchFactory() {        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =                new ConcurrentKafkaListenerContainerFactory<>();        factory.setConsumerFactory(kafkaConsumerFactory());        //设置为批量消费，每个批次数量在Kafka配置参数中设置ConsumerConfig.MAX_POLL_RECORDS_CONFIG        factory.setBatchListener(true);        return factory;    }

复制代码

使用containerFactory = "batchFactory"

clientIdPrefix 客户端前缀

会覆盖消费者工厂的kafka.consumer.client-id属性; 最为前缀后面接 -n n 是数字

concurrency 并发数

会覆盖消费者工厂中的 concurrency ,这里的并发数就是多线程消费; 比如说单机情况下,你设置了 3; 相当于就是启动了 3 个客户端来分配消费分区;分布式情况总线程数=concurrency*机器数量; 并不是设置越多越好,具体如何设置请看 属性concurrency的作用及配置(RoundRobinAssignor 、RangeAssignor)

    /**     * 监听器工厂      * @return     */    @Bean    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> concurrencyFactory() {        ConcurrentKafkaListenerContainerFactory<Integer, String> factory =                new ConcurrentKafkaListenerContainerFactory<>();        factory.setConsumerFactory(kafkaConsumerFactory());        factory.setConcurrency(6);        return factory;    }

复制代码

    @KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3", containerFactory = "concurrencyFactory",concurrency = "1)

复制代码

虽然使用的工厂是concurrencyFactory(concurrency 配置了 6); 但是他最终生成的监听器数量是 1;

properties 配置其他属性

kafka 中的属性看org.apache.kafka.clients.consumer.ConsumerConfig ;同名的都可以修改掉;

用法

    @KafkaListener(id = "consumer-id5",idIsGroup = false,topics = "SHI_TOPIC3", containerFactory = "concurrencyFactory",concurrency = "1"            , clientIdPrefix = "myClientId5",groupId = "groupId-test",            properties = {                    "enable.auto.commit:false","max.poll.interval.ms:6000" },errorHandler="kafkaDefaultListenerErrorHandler")

复制代码

3@KafkaListener 使用

4KafkaListenerEndpointRegistry

    @Autowired    private KafkaListenerEndpointRegistry registry;       //.... 获取所有注册的监听器        registry.getAllListenerContainers();

复制代码

设置入参验证器

当您将 Spring Boot 与验证启动器一起使用时，将 LocalValidatorFactoryBean 自动配置：如下

@Configuration@EnableKafkapublic class Config implements KafkaListenerConfigurer {
    @Autowired    private LocalValidatorFactoryBean validator;    ...
    @Override    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {      registrar.setValidator(this.validator);    }}

复制代码

使用

@KafkaListener(id="validated", topics = "annotated35", errorHandler = "validationErrorHandler",      containerFactory = "kafkaJsonListenerContainerFactory")public void validatedListener(@Payload @Valid ValidatedClass val) {    ...}
@Beanpublic KafkaListenerErrorHandler validationErrorHandler() {    return (m, e) -> {        ...    };}

复制代码

发布于: 2022 年 09 月 15 日阅读数: 50

原文链接:【http://xie.infoq.cn/article/947656b2a29fc29ce16586d0d】。未经作者许可，禁止转载。

石臻臻的杂货铺

关注

关注公众号: 石臻臻的杂货铺获取最新文章 2019.09.06 加入

进高质量滴滴技术交流群,只交流技术不闲聊加 szzdzhp001 进群 20w字《Kafka运维与实战宝典》PDF下载请关注公众号：石臻臻的杂货铺

发布

暂无评论

创作场景

【spring-kafka】@KafkaListener 详解与使用

1 说明

2@KafkaListener 详解

id 监听器的 id

①. 消费者线程命名规则

②.在相同容器中的监听器 ID 不能重复

③.会覆盖消费者工厂的消费组 GroupId

④. 如果配置了属性 groupId,则其优先级最高

groupId 消费组名

如何获取消费者 group.id

topics 指定要监听哪些 topic(与 topicPattern、topicPartitions 三选一)

topicPattern 匹配 Topic 进行监听(与 topics、topicPartitions 三选一)

topicPartitions 显式分区分配

errorHandler 异常处理

containerFactory 监听器工厂

clientIdPrefix 客户端前缀

concurrency 并发数

properties 配置其他属性

3@KafkaListener 使用

4KafkaListenerEndpointRegistry

设置入参验证器

石臻臻的杂货铺

评论