spring-kafka使消费者动态订阅新增的topic问题

 更新时间:2022年12月29日 10:02  点击:448 作者:DayDayUp丶

一、前言

在Java中使用kafka,方式很多,例如:

  • 直接使用kafka-clients这类原生的API;
  • 也可以使用Spring对其的包装API,即spring-kafka,同其它包装API一样(如JdbcTemplate、RestTemplate、RedisTemplate等等),KafkaTemplate是其生产者核心类,KafkaListener是其消费者核心注解;
  • 也有包装地更加抽象的SpringCloudStream等。

这里讨论的话题是,如何在spring-kafka中,使得一个消费者可以动态订阅新增的topic?

本文不讨论利用SpringCloudConfig或Apollo等分布式配置中心,利用@RefreshScope的方式来达到目的,这种方式有点杀鸡用牛刀,也会增加系统复杂度和维护成本。

我的环境:jdk 1.8,Spring 2.1.3.RELEASE,kafka_2.12-2.3.0单节点。

二、需求分析

上面已经提到,spring-kafka通过 @KafkaListener 的方式配置订阅的topic,最常用的属性可能是 topics,而要实现本文的需求,就要使用另一个属性 topicPattern,查看它的属性说明:

The topic pattern for this listener. 
The entries can be 'topic pattern', a'property-placeholder key' or an 'expression'. 
The framework will create acontainer that subscribes to all topics matching the specified pattern to getdynamically assigned partitions. 
The pattern matching will be performedperiodically against topics existing at the time of check. 
An expression mustbe resolved to the topic pattern (String or Pattern result types are supported). 

将其翻译过来:

此侦听器的主题模式。条目可以是“主题模式”,“属性占位符键”或“表达式”。
该框架将创建一个容器,该容器订阅与指定模式匹配的所有主题以获取动态分配的分区。
模式匹配将针对检查时存在的主题【定期执行】。
表达式必须解析为主题模式(支持字符串或模式结果类型)。

注意:从说明信息来看,topicPattern 已经可以做到定期检查topic列表,然后将新加入的topic分配至某个消费者。

下面列出消费端的核心测试代码:

@Component
public class SinkConsumer {
    @KafkaListener(topicPattern = "test_topic2.*")
    public void listen2(ConsumerRecord<?, ?> record) throws Exception {
        System.out.printf("topic2.* = %s, offset = %d, value = %s \n", record.topic(), record.offset(), record.value());
    }
}

代码实现很简洁,就是期待我们新增一个符合 topicPattern 的topic后,spring-kafka能否自动为新建的topic分配到此目标消费者。

三、测试运行

3.1 启动消费者服务

配置文件中,spring该配的配,kafka该配的配,接着启动即可。

3.2 新建topic

新建 test_topic2_3,刚创建完不能立刻分配到目标消费者,从 topicPattern 的注释得知spring-kafka会定期扫描topic列表,我们要给它几分钟等待扫描到新topic,并为它成功分配到目标消费者后,再去发送第一条消息(所以可以先去洗个手,此时19:02)。

3.3 等待topic被分配到消费者

洗手期间的控制台日志提示:已为新建的 test_topic2_3 分配到我们的目标消费者,并将offset设置到起始位置0,日志如下:

2019-11-15 19:05:12.958  INFO 7768 --- [ntainer#1-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-3, groupId=test] Revoking previously assigned partitions [test_topic2_2-0, test_topic2_1-0]
2019-11-15 19:05:12.958  INFO 7768 --- [ntainer#1-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked: [test_topic2_2-0, test_topic2_1-0]
2019-11-15 19:05:12.958  INFO 7768 --- [ntainer#1-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-3, groupId=test] (Re-)joining group
2019-11-15 19:05:15.757  INFO 7768 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=test] Attempt to heartbeat failed since group is rebalancing
2019-11-15 19:05:15.761  INFO 7768 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=test] Revoking previously assigned partitions [test_topic-0]
2019-11-15 19:05:15.762  INFO 7768 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions revoked: [test_topic-0]
2019-11-15 19:05:15.762  INFO 7768 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=test] (Re-)joining group
2019-11-15 19:05:16.025  INFO 7768 --- [ntainer#1-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-3, groupId=test] Successfully joined group with generation 6
2019-11-15 19:05:16.025  INFO 7768 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.AbstractCoordinator  : [Consumer clientId=consumer-2, groupId=test] Successfully joined group with generation 6
2019-11-15 19:05:16.026  INFO 7768 --- [ntainer#1-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-3, groupId=test] Setting newly assigned partitions [test_topic2_2-0, test_topic2_3-0, test_topic2_1-0]
2019-11-15 19:05:16.026  INFO 7768 --- [ntainer#0-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=test] Setting newly assigned partitions [test_topic-0]
2019-11-15 19:05:16.028  INFO 7768 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [test_topic-0]
2019-11-15 19:05:16.032  INFO 7768 --- [ntainer#1-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-3, groupId=test] Resetting offset for partition test_topic2_3-0 to offset 0.
2019-11-15 19:05:16.032  INFO 7768 --- [ntainer#1-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : partitions assigned: [test_topic2_2-0, test_topic2_3-0, test_topic2_1-0]

3.4 发送第一条消息

洗手完毕,看到3.3小节里的日志,然后确认成功分配到目标消费者,且offset被设为0之后,发送第一条消息【我是第1个test_topic2_3的消息】,控制台日志打印出此消息信息,代表成功消费:

topic2.* = test_topic2_3, offset = 0, value = {"date":"2019-11-15 19:11:13","msg":"我是第1个test_topic2_3的消息"} 

3.5 注意事项

若不等到offset被设为0之后,过早发送消息,则会在消费端丢失过早发送的消息,并且当spring-kafka自动设置offset的时候,日志提示,offset被设置为1,而不是起始位置0:

INFO o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumer-3, groupId=test] Resetting offset for partition test_topic2_1-0 to offset 1.

在上面的3.1至3.4的整个过程中,可能会日志警告,代表暂时不能为新增的topic分配到目标消费者:

WARN o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-2, groupId=test] The following subscribed topics are not assigned to any members: [test_topic2_3] 

所以只需等待日志提示可以成功分配到目标消费者,且offset被设为0之后,即可发送第一条消息。

总结

以上为个人经验,希望能给大家一个参考,也希望大家多多支持猪先飞。

原文出处:https://blog.csdn.net/songzehao/article/details/103091486

[!--infotagslink--]

相关文章

  • Spring AOP 对象内部方法间的嵌套调用方式

    这篇文章主要介绍了Spring AOP 对象内部方法间的嵌套调用方式,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-08-29
  • Spring Cloud 中@FeignClient注解中的contextId属性详解

    这篇文章主要介绍了Spring Cloud 中@FeignClient注解中的contextId属性详解,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-09-25
  • Springboot如何实现Web系统License授权认证

    这篇文章主要介绍了Springboot如何实现Web系统License授权认证,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-05-28
  • .NET Core下使用Kafka的方法步骤

    这篇文章主要介绍了.NET Core下使用Kafka的方法步骤,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-09-22
  • 如何在Spring WebFlux的任何地方获取Request对象

    这篇文章主要介绍了如何在Spring WebFlux的任何地方获取Request对象,帮助大家更好的理解和使用springboot框架,感兴趣的朋友可以了解下...2021-01-26
  • 详解SpringCloudGateway内存泄漏问题

    这篇文章主要介绍了详解SpringCloudGateway内存泄漏问题,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-07-16
  • Spring为什么不推荐使用@Autowired注解详析

    @Autowired 注解的主要功能就是完成自动注入,使用也非常简单,但这篇文章主要给大家介绍了关于Spring为什么不推荐使用@Autowired注解的相关资料,需要的朋友可以参考下...2021-11-03
  • Springboot如何使用mybatis实现拦截SQL分页

    这篇文章主要介绍了Springboot使用mybatis实现拦截SQL分页,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-06-19
  • SpringMVC文件上传原理及实现过程解析

    这篇文章主要介绍了SpringMVC文件上传原理及实现过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-07-15
  • Spring Data JPA 关键字Exists的用法说明

    这篇文章主要介绍了Spring Data JPA 关键字Exists的用法说明,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教...2021-06-10
  • tomcat启动完成执行 某个方法 定时任务(Spring)操作

    这篇文章主要介绍了tomcat启动完成执行 某个方法 定时任务(Spring)操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-09-25
  • 使用Maven 搭建 Spring MVC 本地部署Tomcat的详细教程

    这篇文章主要介绍了使用Maven 搭建 Spring MVC 本地部署Tomcat,本文通过图文并茂的形式给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-08-16
  • SpringMvc自动装箱及GET请求参数原理解析

    这篇文章主要介绍了SpringMvc自动装箱及GET请求参数原理解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-09-19
  • Java Spring Cloud 负载均衡详解

    这篇文章主要介绍了Spring Cloud负载均衡及远程调用实现详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2021-09-18
  • Springboot使用thymeleaf动态模板实现刷新

    这篇文章主要介绍了Springboot使用thymeleaf动态模板实现刷新,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-08-31
  • SpringMvc获取请求头请求体消息过程解析

    这篇文章主要介绍了SpringMvc获取请求头请求体消息过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-09-17
  • spring boot 使用utf8mb4的操作

    这篇文章主要介绍了spring boot 使用utf8mb4的操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-02-20
  • Springmvc ResponseBody响应json数据实现过程

    这篇文章主要介绍了Springmvc ResponseBody响应json数据实现过程,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-10-26
  • Idea打包springboot项目没有.original文件解决方案

    这篇文章主要介绍了Idea打包springboot项目没有.original文件解决方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-07-26
  • SpringData Repository接口用法解析

    这篇文章主要介绍了SpringData Repository接口用法解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-08-27