kafka如何动态消费新增topic主题-Toy模板网

这篇具有很好参考价值的文章主要介绍了kafka如何动态消费新增topic主题。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、解决痛点

使用spring-kafka客户端，每次新增topic主题，都需要硬编码客户端并重新发布服务，操作麻烦耗时长。kafkaListener虽可以支持通配符消费topic，缺点是并发数需要手动改并且重启服务。对于业务逻辑相似场景，创建新主题动态监听可以用kafka-batch-starter组件

二、组件能力

1、新增topic名称为：auto.topic1（由于配置spring.kafka.consumer.prefix 为auto，因此只有auto前缀的topic，才会被组件动态监听。）

2、应用输出日志，监听到新增auto.topic1，并初始化客户端 (主题刷新间隔为10s)

3、发新的消息给auto.topic1，可以看到消费成功，不影响原有kafka客户端的消费

接入成功

三、核心设计点

1、动态接入消息，无需重新发布服务

2、适配spring-kafka配置，和生产运行中的spring-kafka客户端不冲突

四、接入步骤

1、引入maven依赖坐标

<dependency>
    <groupId>io.gitee.qrkcn</groupId>
    <artifactId>kafka-batch-starter</artifactId>
    <version>1.0.0</version>
</dependency>

2、客户端配置，适配spring-kafka原有的配置文件

spring:
  kafka:
    bootstrap-servers: 172.111.0.1:9092
    consumer:
      prefix: auto
      group-id: kafka-local1
      # 手动提交
      enable-auto-commit: false
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        auto.commit.interval.ms: 1000
        session.timeout.ms: 60000
        request.timeout.ms: 1000000
        max.poll.interval.ms: 300000   #默认5分钟 300000ms
        max.poll.records: 30   #默认500条

spring.kafka.consumer.prefix 配置说明：

必填，topic名称前缀符合配置，才会动态监听

不填为null，组件抛出自定义异常：Caused by: java.lang.InterruptedException: spring.kafka.consumer.prefix can not be null if use kafka-batch-starter

填入""空字符串，则会动态监听所有主题

3、实现动态消费方法DynamicConsumerMethod

在processMessage中处理业务逻辑文章来源地址https://www.toymoban.com/news/detail-420278.html

@Component
@Slf4j
public class DynamicConsumerMethodImpl implements DynamicConsumerMethod {
    @Override
    public void processMessage(ConsumerRecords<String, String> records, String topicName) {
        for (ConsumerRecord<String, String> record : records) {
            String key = record.key();
            String value = record.value();
            int partition = record.partition();
            String topic = record.topic();
            log.info("consumer.topicName={},key={},value={}",topicName, key,value);
        }
    }
}

到了这里，关于kafka如何动态消费新增topic主题的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！