springboot+kafka中@KafkaListener如何动态指定多个topic

不过我们可以有个折中的办法，就是利用@KafkaListener的topicPattern参数来进行topic匹配。
具体如何操作的可以看下这位老哥的blog:
https://blog.csdn.net/songzehao/article/details/103091486

终极方法 ：

1. 不使用@KafkaListener，使用kafka原生客户端依赖，手动初始化消费者，开启消费者线程。
2. 在消费者线程中，每次循环都从配置、数据库或者其他配置源获取最新的topic信息，与之前的topic比较，如果发生变化，重新订阅topic或者初始化消费者。

1. 加入kafka客户端依赖(本次测试服务端kafka版本：2.12-2.4.0)

   <dependency>
   	<groupId>org.apache.kafka</groupId>
   	<artifactId>kafka-clients</artifactId>
   	<version>2.3.0</version>
   </dependency>
   

2. 代码

   @Service
   @Slf4j
   public class KafkaConsumers implements InitializingBean {
        * 消费者
       private static KafkaConsumer<String, String> consumer;
        * topic
       private List<String> topicList;
       public static String getNewTopic() {
           try {
               return org.apache.commons.io.FileUtils.readLines(new File("D:/topic.txt"), "utf-8").get(0);
           } catch (IOException e) {
               e.printStackTrace();
           return null;
        * 初始化消费者（配置写死是为了快速测试，请大家使用配置文件）
        * @param topicList
        * @return
       public KafkaConsumer<String, String> getInitConsumer(List<String> topicList) {
           //配置信息
           Properties props = new Properties();
           //kafka服务器地址
           props.put("bootstrap.servers", "192.168.9.185:9092");
           //必须指定消费者组
           props.put("group.id", "haha");
           //设置数据key和value的序列化处理类
           props.put("key.deserializer", StringDeserializer.class);
           props.put("value.deserializer", StringDeserializer.class);
           //创建消息者实例
           KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
           //订阅topic的消息
           consumer.subscribe(topicList);
           return consumer;
        * 开启消费者线程
        * 异常请自己根据需求自己处理
       @Override
       public void afterPropertiesSet() {
           // 初始化topic
           topicList = Splitter.on(",").splitToList(Objects.requireNonNull(getNewTopic()));
           if (org.apache.commons.collections.CollectionUtils.isNotEmpty(topicList)) {
               consumer = getInitConsumer(topicList);
               // 开启一个消费者线程
               new Thread(() -> {
                   while (true) {
                       // 模拟从配置源中获取最新的topic（字符串，逗号隔开）
                       final List<String> newTopic = Splitter.on(",").splitToList(Objects.requireNonNull(getNewTopic()));
                       // 如果topic发生变化
                       if (!topicList.equals(newTopic)) {
                           log.info("topic 发生变化：newTopic:{},oldTopic:{}-------------------------", newTopic, topicList);
                           // method one：重新订阅topic:
                           topicList = newTopic;
                           consumer.subscribe(newTopic);
                           // method two：关闭原来的消费者，重新初始化一个消费者
                           //consumer.close();
                           //topicList = newTopic;
                           //consumer = getInitConsumer(newTopic);
                           continue;
                       ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                       for (ConsumerRecord<String, String> record : records) {
                           System.out.println("key:" + record.key() + "" + ",value:" + record.value());
               }).start();
   

   说一下第72行代码：

   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   

   上面这行代码表示：在100ms内等待Kafka的broker返回数据.超市参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回。

   在修改topic后，必须等到此次poll拉取的消息处理完，while(true)循环的时候检测topic发生变化，才能重新订阅topic.
   poll()方法一次拉取得消息数默认为：500，如下图，kafka客户端源码中设置的。
   

   如果想自定义此配置，可在初始化消费者时加入
   

3. 运行结果(测试的topic中都无数据)
   
   注意：KafkaConsumer是线程不安全的，不要用一个KafkaConsumer实例开启多个消费者，要开启多个消费者，需要new 多个KafkaConsumer实例。

   有什么问题欢迎提出来讨论和交流 q:281867465