【spring-kafka】@KafkaListener 详解与使用
作者:石臻臻,CSDN 博客之星 Top5、Kafka Contributor、nacos Contributor、华为云 MVP,腾讯云 TVP,滴滴 Kafka 技术专家、 KnowStreaming PMC)。
KnowStreaming 是滴滴开源的Kafka运维管控平台, 有兴趣一起参与参与开发的同学,但是怕自己能力不够的同学,可以联系我,带你一起你参与开源! 。
Kafka 高质量专栏请看 石臻臻的杂货铺的Kafka专栏
1 说明
从 2.2.4 版开始,您可以直接在注释上指定 Kafka 使用者属性,这些属性将覆盖在使用者工厂中配置的具有相同名称的所有属性。您不能通过这种方式指定 group.id 和 client.id 属性。他们将被忽略;
可以使用 #{…}或属性占位符(${…})在 SpEL 上配置注释上的大多数属性。比如:
属性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 不能重复
否则会报错
③.会覆盖消费者工厂的消费组 GroupId
假如配置文件属性配置了消费组kafka.consumer.group-id=BASE-DEMO
正常情况它是该容器中的默认消费组但是如果设置了 @KafkaListener(id = "consumer-id7", topics = {"SHI_TOPIC3"})
那么当前消费者的消费组就是consumer-id7
;
当然如果你不想要他作为 groupId 的话 可以设置属性idIsGroup = false
;那么还是会使用默认的 GroupId;
④. 如果配置了属性 groupId,则其优先级最高
例如上面代码中最终这个消费者的消费组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 显式分区分配
可以为监听器配置明确的主题和分区(以及可选的初始偏移量)
上面例子意思是 监听topic1
的 0,1 分区;监听topic2
的第 0 分区,并且第 1 分区从 offset 为 100 的开始消费;
errorHandler 异常处理
实现KafkaListenerErrorHandler
; 然后做一些异常处理;
调用的时候 填写 beanName;例如errorHandler="kafkaDefaultListenerErrorHandler"
containerFactory 监听器工厂
指定生成监听器的工厂类;
例如我写一个 批量消费的工厂类
使用containerFactory = "batchFactory"
clientIdPrefix 客户端前缀
会覆盖消费者工厂的
kafka.consumer.client-id
属性; 最为前缀后面接-n
n 是数字
concurrency 并发数
会覆盖消费者工厂中的 concurrency ,这里的并发数就是多线程消费; 比如说单机情况下,你设置了 3; 相当于就是启动了 3 个客户端来分配消费分区;分布式情况 总线程数=concurrency*机器数量; 并不是设置越多越好,具体如何设置请看 属性concurrency的作用及配置(RoundRobinAssignor 、RangeAssignor)
虽然使用的工厂是concurrencyFactory
(concurrency 配置了 6); 但是他最终生成的监听器数量 是 1;
properties 配置其他属性
kafka 中的属性看org.apache.kafka.clients.consumer.ConsumerConfig
;同名的都可以修改掉;
用法
3@KafkaListener 使用
4KafkaListenerEndpointRegistry
设置入参验证器
当您将 Spring Boot 与验证启动器一起使用时,将 LocalValidatorFactoryBean 自动配置:如下
使用
TODO..
![](https://static001.geekbang.org/infoq/3f/3f2956dffa55898256baca659a8107c2.png)
在这里插入图片描述
版权声明: 本文为 InfoQ 作者【石臻臻的杂货铺】的原创文章。
原文链接:【http://xie.infoq.cn/article/e21b18d64c2a0160312bac7ed】。未经作者许可,禁止转载。
评论