搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符

2024-04-30 20:36
文章标签 java springboot spring 搭建 整合 多个 取消 数据源 限定符 kafka 分布式服务 三十七 大型

本文主要是介绍搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

系列文章目录

文章目录

  • 系列文章目录
  • 前言
      • 一、本文要点
      • 二、开发环境
      • 三、原项目
      • 四、修改项目
      • 五、测试一下
      • 五、小结

前言

本插件稳定运行上百个kafka项目，每天处理上亿级的数据的精简小插件，快速上手。

<dependency><groupId>io.github.vipjoey</groupId><artifactId>multi-kafka-consumer-starter</artifactId><version>最新版本号</version>
</dependency>

例如下面这样简单的配置就完成SpringBoot和kafka的整合，我们只需要关心com.mmc.multi.kafka.starter.OneProcessor和com.mmc.multi.kafka.starter.TwoProcessor 这两个Service的代码开发。

## topic1的kafka配置
spring.kafka.one.enabled=true
spring.kafka.one.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.one.topic=mmc-topic-one
spring.kafka.one.group-id=group-consumer-one
spring.kafka.one.processor=com.mmc.multi.kafka.starter.OneProcessor // 业务处理类名称
spring.kafka.one.consumer.auto-offset-reset=latest
spring.kafka.one.consumer.max-poll-records=10
spring.kafka.one.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.one.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer## topic2的kafka配置
spring.kafka.two.enabled=true
spring.kafka.two.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.two.topic=mmc-topic-two
spring.kafka.two.group-id=group-consumer-two
spring.kafka.two.processor=com.mmc.multi.kafka.starter.TwoProcessor // 业务处理类名称
spring.kafka.two.consumer.auto-offset-reset=latest
spring.kafka.two.consumer.max-poll-records=10
spring.kafka.two.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.two.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

国籍惯例，先上源码：Github源码

一、本文要点

本文将介绍通过封装一个starter，来实现多kafka数据源的配置，通过通过源码，可以学习以下特性。系列文章完整目录

  • SpringBoot 整合多个kafka数据源
  • SpringBoot 批量消费kafka消息
  • SpringBoot 优雅地启动或停止消费kafka
  • SpringBoot kafka本地单元测试（免集群）
  • SpringBoot 利用map注入多份配置
  • SpringBoot BeanPostProcessor 后置处理器使用方式
  • SpringBoot 将自定义类注册到IOC容器
  • SpringBoot 注入bean到自定义类成员变量
  • Springboot 取消限定符

二、开发环境

  • jdk 1.8
  • maven 3.6.2
  • springboot 2.4.3
  • kafka-client 2.6.6
  • idea 2020

三、原项目

1、接前文，我们开发了一个kafka插件，但在使用过程中发现有些不方便的地方，例如我们所有processor需要继承MmcKafkaKafkaAbastrctProcessor<T extends MmcKafkaMsg> ，其中的T为反序列化的实体类类型。


@Slf4j
@Service
public class OneProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {@Resourceprivate DemoService demoService;@Overrideprotected Class<DemoMsg> getEntityClass() {return DemoMsg.class;}@Overrideprotected void dealMessage(List<DemoMsg> datas) {demoService.dealMessage("one", datas.stream().map(x -> (MmcKafkaMsg) x).collect(Collectors.toList()));}}@Slf4j
@Service
public class TwoProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {@Resourceprivate DemoService demoService;public TwoProcessor() {}@Overrideprotected Class<DemoMsg> getEntityClass() {return DemoMsg.class;}@Overrideprotected void dealMessage(List<DemoMsg> datas) {demoService.dealMessage("two", datas.stream().map(x -> (MmcKafkaMsg) x).collect(Collectors.toList()));}}

2、可以看到这里有两个体验不太好的地方。

  • 自定义实体类DemoMsg 必须要继承 MmcKafkaMsg，很多同学会忘记这个步骤；
  • 需要覆盖getEntityClass()父类方法，用于反序列化指定实体类的类型，这里太冗余；

因此、所以我们要升级和优化。

四、修改项目

1、取消限定符，消息实体类不再强制要求实现MmcKafkaMsg接口，改为可选项，作为候选插件化的能力增强（后文介绍）；

@Data
class DemoMsg {private String routekey;private String name;private Long timestamp;}

2、修改MmcKafkaKafkaAbastrctProcessor类，取消限定符并增加类型推断方法。

a、如果实现了MmcKafkaMsg接口，就拥有了单次消费内的batch数据去重能力；

public void onMessage(List<ConsumerRecord<String, String>> records) {if (null == records || CollectionUtils.isEmpty(records)) {log.warn("{} records is null or records.value is empty.", name);return;}Assert.hasText(name, "You must pass the field `name` to the Constructor or invoke the setName() after the class was created.");Assert.notNull(properties, "You must pass the field `properties` to the Constructor or invoke the setProperties() after the class was created.");try {Stream<T> dataStream = records.stream().map(ConsumerRecord::value).flatMap(this::doParse).filter(Objects::nonNull).filter(this::isRightRecord);// 支持配置强制去重或实现了接口能力去重if (properties.isDuplicate() || isSubtypeOfInterface(MmcKafkaMsg.class)) {// 检查是否实现了去重接口if (!isSubtypeOfInterface(MmcKafkaMsg.class)) {throw new RuntimeException("The interface "+ MmcKafkaMsg.class.getName() + " is not implemented if you set the config `spring.kafka.xxx.duplicate=true` .");}dataStream = dataStream.collect(Collectors.groupingBy(this::buildRoutekey)).entrySet().stream().map(this::findLasted).filter(Objects::nonNull);}List<T> datas = dataStream.collect(Collectors.toList());if (CommonUtil.isNotEmpty(datas)) {this.dealMessage(datas);}} catch (Exception e) {log.error(name + "-dealMessage error ", e);}}

b、新增类型推断方法，目的是去掉子类必须实现getEntityClass()的约束；

    protected boolean isSubtypeOfInterface(Class<?> interfaceClass) {if (null == type) {Type superClass = getClass().getGenericSuperclass();if (superClass instanceof ParameterizedType) {ParameterizedType parameterizedType = (ParameterizedType) superClass;Type[] typeArguments = parameterizedType.getActualTypeArguments();if (typeArguments.length > 0 && typeArguments[0] instanceof Class) {//noinspection uncheckedtype = (Class<T>) typeArguments[0];}}}return (null != type) && interfaceClass.isAssignableFrom(type);}protected  Class<T> getEntityClass() {if (null == type) {synchronized(this) {Type superClass = getClass().getGenericSuperclass();if (superClass instanceof ParameterizedType) {ParameterizedType parameterizedType = (ParameterizedType) superClass;Type[] typeArguments = parameterizedType.getActualTypeArguments();if (typeArguments.length > 0 && typeArguments[0] instanceof Class) {//noinspection uncheckedtype = (Class<T>) typeArguments[0];}}}}return type;}

c、修改去重方法，也就是取批次内最新一条消息，不再使用限定符；

    protected T findLasted(Map.Entry<String, List<T>> entry) {try {Optional<T> d = entry.getValue().stream().max(Comparator.comparing(x -> ((PandoKafkaMsg) x).getRoutekey()));if (d.isPresent()) {return d.get();}} catch (Exception e) {String content = JsonUtil.toJsonStr(entry.getValue());log.error("处理消息出错:{}", e.getMessage() + ": " + content, e);}return null;}protected String buildRoutekey(T t) {return ((MmcKafkaMsg) t).getRoutekey();}

3、修改MmcKafkaBeanPostProcessor，取消限定符。

public class MmcKafkaBeanPostProcessor implements BeanPostProcessor {@Getterprivate final Map<String, MmcKafkaKafkaAbastrctProcessor<?>> suitableClass = new ConcurrentHashMap<>();@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {if (bean instanceof MmcKafkaKafkaAbastrctProcessor) {MmcKafkaKafkaAbastrctProcessor<?> target = (MmcKafkaKafkaAbastrctProcessor<?>) bean;suitableClass.putIfAbsent(beanName, target);suitableClass.putIfAbsent(bean.getClass().getName(), target);}return bean;}
}

4、修改MmcKafkaProcessorFactory，取消限定符。

五、测试一下

1、引入kafka测试需要的jar。参考文章：kafka单元测试

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><scope>test</scope></dependency>

2、定义一个消息实体和业务处理类。

@Data
class DemoMsg  {private String routekey;private String name;private Long timestamp;}
@Slf4j
@Service
public class OneProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {@Resourceprivate DemoService demoService;@Overrideprotected void dealMessage(List<DemoMsg> datas) {datas.forEach(x -> {log.info("dealMessage one: {}", x);});}}

3、配置kafka地址和指定业务处理类。

spring.kafka.one.enabled=true
spring.kafka.one.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.one.topic=mmc-topic-one
spring.kafka.one.group-id=group-consumer-one
spring.kafka.one.processor=com.mmc.multi.kafka.starter.OneProcessor  // 业务处理类名称
spring.kafka.one.consumer.auto-offset-reset=latest
spring.kafka.one.consumer.max-poll-records=10
spring.kafka.one.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.one.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer

4、编写测试类。

@Slf4j
@ActiveProfiles("dev")
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = {MmcMultiConsumerAutoConfiguration.class, DemoService.class, OneProcessor.class})
@TestPropertySource(value = "classpath:application.properties")
@DirtiesContext
@EmbeddedKafka(topics = {"${spring.kafka.one.topic}"})
class AppTest {@Resourceprivate EmbeddedKafkaBroker embeddedKafkaBroker;@Value("${spring.kafka.one.topic}")private String topicOne;@Value("${spring.kafka.two.topic}")private String topicTwo;@Testvoid testDealMessage() throws Exception {// 模拟生产数据produceMessage();Thread.sleep(10 * 1000);}void produceMessage() {Map<String, Object> configs = new HashMap<>(KafkaTestUtils.producerProps(embeddedKafkaBroker));Producer<String, String> producer = new DefaultKafkaProducerFactory<>(configs, new StringSerializer(), new StringSerializer()).createProducer();for (int i = 0; i < 10; i++) {DemoMsg msg = new DemoMsg();msg.setRoutekey("routekey" + i);msg.setName("name" + i);msg.setTimestamp(System.currentTimeMillis());String json = JsonUtil.toJsonStr(msg);producer.send(new ProducerRecord<>(topicOne, "my-aggregate-id", json));producer.send(new ProducerRecord<>(topicTwo, "my-aggregate-id", json));producer.flush();}}
}

5、运行一下，测试通过。
在这里插入图片描述

五、小结

将本项目代码构建成starter，就可以大大提升我们开发效率，我们只需要关心业务代码的开发，github项目源码：轻触这里。如果对你有用可以打个星星哦。下一篇，升级本starter，在kafka单分区下实现十万级消费处理速度。

加我加群一起交流学习！更多干货下载、项目源码和大厂内推等着你

这篇关于搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

原文地址:
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.chinasem.cn/article/949852。

相关文章

深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

深入浅出Spring中的@Autowired自动注入的工作原理及实践应用

《深入浅出Spring中的@Autowired自动注入的工作原理及实践应用》在Spring框架的学习旅程中,@Autowired无疑是一个高频出现却又让初学者头疼的注解,它看似简单,却蕴含着Sprin... 目录深入浅出Spring中的@Autowired：自动注入的奥秘什么是依赖注入？@Autowired
阅读更多...
Spring 依赖注入与循环依赖总结

Spring 依赖注入与循环依赖总结

《Spring依赖注入与循环依赖总结》这篇文章给大家介绍Spring依赖注入与循环依赖总结篇,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录1. Spring 三级缓存解决循环依赖1. 创建UserService原始对象2. 将原始对象包装成工
阅读更多...
Java中如何正确的停掉线程

Java中如何正确的停掉线程

《Java中如何正确的停掉线程》Java通过interrupt()通知线程停止而非强制,确保线程自主处理中断,避免数据损坏,线程池的shutdown()等待任务完成,shutdownNow()强制中断... 目录为什么不强制停止为什么 Java 不提供强制停止线程的能力呢？如何用interrupt停止线程s
阅读更多...
SpringBoot请求参数传递与接收示例详解

SpringBoot请求参数传递与接收示例详解

《SpringBoot请求参数传递与接收示例详解》本文给大家介绍SpringBoot请求参数传递与接收示例详解,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋... 目录I. 基础参数传递i.查询参数（Query Parameters）ii.路径参数（Path Va
阅读更多...
SpringBoot路径映射配置的实现步骤

SpringBoot路径映射配置的实现步骤

《SpringBoot路径映射配置的实现步骤》本文介绍了如何在SpringBoot项目中配置路径映射,使得除static目录外的资源可被访问,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一... 目录SpringBoot路径映射补：springboot 配置虚拟路径映射 @RequestMapp
阅读更多...
Java MCP 的鉴权深度解析

Java MCP 的鉴权深度解析

《JavaMCP的鉴权深度解析》文章介绍JavaMCP鉴权的实现方式,指出客户端可通过queryString、header或env传递鉴权信息,服务器端支持工具单独鉴权、过滤器集中鉴权及启动时鉴权... 目录一、MCP Client 侧（负责传递，比较简单）（1）常见的 mcpServers json 配置
阅读更多...
GSON框架下将百度天气JSON数据转JavaBean

GSON框架下将百度天气JSON数据转JavaBean

《GSON框架下将百度天气JSON数据转JavaBean》这篇文章主要为大家详细介绍了如何在GSON框架下实现将百度天气JSON数据转JavaBean,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下... 目录前言一、百度天气jsON1、请求参数2、返回参数3、属性映射二、GSON属性映射实战1、类对象映
阅读更多...
Java Stream 并行流简介、使用与注意事项小结

Java Stream 并行流简介、使用与注意事项小结

《JavaStream并行流简介、使用与注意事项小结》Java8并行流基于StreamAPI,利用多核CPU提升计算密集型任务效率,但需注意线程安全、顺序不确定及线程池管理,可通过自定义线程池与C... 目录1. 并行流简介​特点：​2. 并行流的简单使用​示例：并行流的基本使用​3. 配合自定义线程池​示
阅读更多...
从原理到实战解析Java Stream 的并行流性能优化

从原理到实战解析Java Stream 的并行流性能优化

《从原理到实战解析JavaStream的并行流性能优化》本文给大家介绍JavaStream的并行流性能优化：从原理到实战的全攻略,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的... 目录一、并行流的核心原理与适用场景二、性能优化的核心策略1. 合理设置并行度：打破默认阈值2. 避免装箱
阅读更多...
解决升级JDK报错:module java.base does not“opens java.lang.reflect“to unnamed module问题

解决升级JDK报错:module java.base does not“opens java.lang.reflect“to unnamed module问题

《解决升级JDK报错:modulejava.basedoesnot“opensjava.lang.reflect“tounnamedmodule问题》SpringBoot启动错误源于Jav... 目录问题描述原因分析解决方案总结问题描述启动sprintboot时报以下错误原因分析编程异js常是由Ja
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2