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

@Service
@Slf4j
public class KafkaConsumers implements InitializingBean {

    /**
     * 消費(fèi)者
     */
    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;
    }

    /**
     * 初始化消費(fèi)者（配置寫死是為了快速測(cè)試，請(qǐng)大家使用配置文件）
     *
     * @param topicList
     * @return
     */
    public KafkaConsumer<String, String> getInitConsumer(List<String> topicList) {
        //配置信息
        Properties props = new Properties();
        //kafka服務(wù)器地址
        props.put("bootstrap.servers", "192.168.9.185:9092");
        //必須指定消費(fèi)者組
        props.put("group.id", "haha");
        //設(shè)置數(shù)據(jù)key和value的序列化處理類
        props.put("key.deserializer", StringDeserializer.class);
        props.put("value.deserializer", StringDeserializer.class);
        //創(chuàng)建消息者實(shí)例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //訂閱topic的消息
        consumer.subscribe(topicList);
        return consumer;
    }

    /**
     * 開啟消費(fèi)者線程
     * 異常請(qǐng)自己根據(jù)需求自己處理
     */
    @Override
    public void afterPropertiesSet() {
        // 初始化topic
        topicList = Splitter.on(",").splitToList(Objects.requireNonNull(getNewTopic()));
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(topicList)) {
            consumer = getInitConsumer(topicList);
            // 開啟一個(gè)消費(fèi)者線程
            new Thread(() -> {
                while (true) {
                    // 模擬從配置源中獲取最新的topic（字符串，逗號(hào)隔開）
                    final List<String> newTopic = Splitter.on(",").splitToList(Objects.requireNonNull(getNewTopic()));
                    // 如果topic發(fā)生變化
                    if (!topicList.equals(newTopic)) {
                        log.info("topic 發(fā)生變化：newTopic:{},oldTopic:{}-------------------------", newTopic, topicList);
                        // method one：重新訂閱topic:
                        topicList = newTopic;
                        consumer.subscribe(newTopic);
                        // method two：關(guān)閉原來(lái)的消費(fèi)者，重新初始化一個(gè)消費(fèi)者
                        //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();
        }
    }
}

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