搜索
简帛阁>技术文章>kafka中的@KafkaListener如何动态获得topic

kafka中的@KafkaListener如何动态获得topic

第一种方法:
我的天啊,经过几天奋战,终于解决这个大坑,我一定要写个博客纪念一下。
首先,引入依赖:

org.springframework.kafka
spring-kafka

很重要,一定要看好,依赖困扰了我很久
其次写一个 KafkaTopicConfig类,继承InitializingBean,重写afterPropertiesSet()方法。
我的如下:

@Configuration
public class KafkaTopicConfig implements InitializingBean {
@Override
public void afterPropertiesSet() {
//获取topic
    String topics = Sets.newHashSet(KafkaTopicEnum.values()).stream()
            .map(KafkaTopicEnum::getTopic).collect(Collectors.joining(","));
 //系统写入
    System.setProperty("topics", topics);
 }
}

记得配置文件中加上 topics:
然后
@KafkaListener(topics = “#{’${topics}’.split(’,’)}”)
在要调用 @KafkaListener的类前加上@DependsOn(value = “kafkaTopicConfig”),确保kafkaTopicConfig类在此之前加载。
不然还会报错。

注意:@KafkaLisener中的topics是string[]类型,一定要注意传入参数的属性,不然会报value '[Ljava.lang.String;@1fb8997’的错误,一定要注意一下!
最后,感谢网上大佬们给我的不尽的帮助,我太难了啊!!!

第二种方法

KafkaListener
KafkaListener就是kafka消息的消费者
properties+clientIdPrefix用来指定Kafka Consumer的属性
id+topics+topicPattern+topicPartitions+groupId+idIsGroup指定了用什么groupid,从哪里消费,从哪个topic的哪个partition的哪个offset开始读;
containerFactory指定了谁来处理跟kafka的对接,autoStartup + concurrency指定了对接器的两个行为;
beanRef 指定一个存放上述信息的java bean

@KafkaListener(beanRef = "__x", topics = "#{__x.topic}",
 groupId = "#{__x.topic}.group")
一、定时启动的应用场景比如现在单机环境下,我们需要利用Kafka做数据持久化的功能,由于用户活跃的时间为早上10点至晚上12点,那在这个时间段做一个大数据量的持久化可能会影响数据库性能导致用户体验降
应用服务启动时,KafkaListener默认会自动启动进行消费,如果想不自动消费,可以设置AutoStartup属性值为false@Override@KafkaListener(idConsume
一、KafkaListener消费/***手动提交监听**@paramrecord消息记录*@paramack确认实例*/@Override@KafkaListener(idConsumerConst
Kafka创建topic命令很简单,一条命令足矣:bin/kafkatopicsshcreatezookeeperlocalhost:2181replicationfactor3partitions3
Kafka创建topic命令很简单,一条命令足矣:bin/kafkatopicsshcreatezookeeperlocalhost:2181replicationfactor3partitions3
目录写在前面命令行模式代码模式总结写在前面众所周知,kafka是现代流行的消息队列,它使用经典的消息订阅发布模式实现消息的流转,大部分代码结合kafka使用都是使用它的生产者和消费者来实现消息的传递,
logretention{ms,minutes,hours}:Kafkasegmentlog的保存周期,默认保存7天1把retentionms改小,如1000ms,即1分钟/kafkaconfigss
我们在使用kafka时,初期创建时所指定topic属性需要修改,如何动态修改kafka属性?kafka提供了命令行工具—kafkatopicsshkafkatopicssh工具介绍kafkatopic
众所周知,由于Zookeeper并不适合大批量的频繁写入操作,新版Kafka已推荐将consumer的位移信息保存在Kafka内部的topic中,即__consumer_offsetstopic,并且
复“资源”领取独家整理的学习资料!每一个成功人士的背后,必定曾经做出过勇敢而又孤独的决定。放弃不难,但坚持很酷~版本:kafka_211110本文提供两种方式来查看消费者组的消费情况,分别通过命令行