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; 然后做一些异常处理;

@Component

public 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”)

自我介绍一下,小编13年上海交大毕业,曾经在小公司待过,也去过华为、OPPO等大厂,18年进入阿里一直到现在。

深知大多数Java工程师,想要提升技能,往往是自己摸索成长或者是报班学习,但对于培训机构动则几千的学费,着实压力不小。自己不成体系的自学效果低效又漫长,而且极易碰到天花板技术停滞不前!

因此收集整理了一份《2024年Java开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友,同时减轻大家的负担。img

既有适合小白学习的零基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!

由于文件比较大,这里只是将部分目录截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且会持续更新!

如果你觉得这些内容对你有帮助,可以扫码获取!!(备注Java获取)

img

Java高频面试专题合集解析:

阿里Java岗面试百题:Spring 缓存 JVM 微服务 数据库 RabbitMQ等

当然在这还有更多整理总结的Java进阶学习笔记和面试题未展示,其中囊括了Dubbo、Redis、Netty、zookeeper、Spring cloud、分布式、高并发等架构资料和完整的Java架构学习进阶导图!

阿里Java岗面试百题:Spring 缓存 JVM 微服务 数据库 RabbitMQ等

更多Java架构进阶资料展示

阿里Java岗面试百题:Spring 缓存 JVM 微服务 数据库 RabbitMQ等

阿里Java岗面试百题:Spring 缓存 JVM 微服务 数据库 RabbitMQ等

阿里Java岗面试百题:Spring 缓存 JVM 微服务 数据库 RabbitMQ等
《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》点击传送门即可获取!
、Spring cloud、分布式、高并发等架构资料和完整的Java架构学习进阶导图!**

[外链图片转存中…(img-WLOJbfTM-1713405414697)]

更多Java架构进阶资料展示

[外链图片转存中…(img-NQKoN7YW-1713405414698)]

[外链图片转存中…(img-EpAFNmg5-1713405414698)]

[外链图片转存中…(img-6yBqh9yh-1713405414698)]
《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》点击传送门即可获取!

Logo

魔乐社区(Modelers.cn) 是一个中立、公益的人工智能社区,提供人工智能工具、模型、数据的托管、展示与应用协同服务,为人工智能开发及爱好者搭建开放的学习交流平台。社区通过理事会方式运作,由全产业链共同建设、共同运营、共同享有,推动国产AI生态繁荣发展。

更多推荐