搭建大型分布式服务（四十四）SpringBoot 无代码侵入实现多Kafka数据源：单分区提升至十万级消费速度！

2024-09-02 11:52
文章标签 java 代码 实现 springboot spring 搭建 提升 四十四 分区 数据源 速度 kafka 分布式服务 大型 消费 十万 侵入

本文主要是介绍搭建大型分布式服务（四十四）SpringBoot 无代码侵入实现多Kafka数据源：单分区提升至十万级消费速度！，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

系列文章目录

文章目录

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

前言

在过去的一段时间里，我们利用了AI大模型写了一个多线程并发框架，那么，我们怎样集成到Kafka组件里，让消费速度提升N倍呢？

  • 《AI大模型编写多线程并发框架（六十一）：从零开始搭建框架》
  • 《AI大模型编写多线程并发框架（六十二）：限流和并发度优化》
  • 《AI大模型编写多线程并发框架（六十三）：监听器优化·上》
  • 《AI大模型编写多线程并发框架（六十四）：监听器优化·下》
  • 《AI大模型编写多线程并发框架（六十五）：发布和应用》

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

一、本文要点

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

  • SpringBoot 整合多个kafka数据源
  • SpringBoot 批量消费kafka消息
  • SpringBoot 优雅地启动或停止消费kafka
  • SpringBoot kafka本地单元测试（免集群）
  • SpringBoot 利用map注入多份配置
  • SpringBoot BeanPostProcessor 后置处理器使用方式
  • SpringBoot 将自定义类注册到IOC容器
  • SpringBoot 注入bean到自定义类成员变量
  • Springboot 取消限定符
  • SpringBoot 支持消费protobuf类型的kafka消息
  • SpringBoot Aware设计模式
  • SpringBoot 获取kafka消息中的topic、offset、partition、header等参数
  • SpringBoot 使用任意生产者发送kafka消息
  • SpringBoot 配置任意数量的kafka生产者
  • SpringBoot Kafka单次batch消息内并发处理

二、开发环境

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

三、原项目

1、接前文，我们已经发布了Kafka组件到中央仓库，所有开发者都可以下载使用本组件。虽然本组件支持批量消费Kafka消息，但是毕竟它是串行顺序处理的，尤其涉及高IO耗时调用时，比如消费Kafka，然后读写DB多表操作这种场景，会使消费速度下降。能否并发处理这些Kafka消息呢？

答案是可以的、但我们要升级和优化一下。

四、修改项目

1、新增ContainerConfig接口类，用于获取多线程任务容器配置，便于后续使用Apollo、Disconf、Consul等配置中心。

public interface ContainerConfig {/*** Get the execute rate.** @return rate*/int getRate();/*** Get the max task count for per thread.** @return max count*/int getThreshold();/*** The max thread count, default is numbers of processor.* @return count*/default int getParallelism() {return Runtime.getRuntime().availableProcessors();}
}

2、修改MmcMultiKafkaProperties配置类，增加容器配置；

@ToString
@Data
@ConfigurationProperties(prefix = "spring")
public class MmcMultiKafkaProperties {// 省略其他代码/*** MmcKafkaProperties.*/@Datastatic class MmcKafkaProperties {// 省略其他代码/*** 并发设置.*/private Container container = new Container();}@Datapublic static class Container implements ContainerConfig {/*** 是否启用多线程消费.*/private boolean enabled = true;/** 消费消息的速率（每秒接收的记录数），默认值为1000.*/private int rate = 1000;/** 最小批次数量，默认为2.*/private int threshold = 2;/** 设置并行度，默认值为可用处理器数量.*/private int parallelism = Runtime.getRuntime().availableProcessors();}
}

3、修改MmcMultiConsumerAutoConfiguration配置类，主要是增加inputer的初始化方法，用于后续构建多线程任务容器实例。


public interface MmcInputer {// 省略其他代码/*** 初始化kafka容器.*/void init();
}@Slf4j
@Configuration
@EnableConfigurationProperties(MmcMultiKafkaProperties.class)
@ConditionalOnProperty(prefix = "spring.kafka", value = "enabled", matchIfMissing = true)
public class MmcMultiConsumerAutoConfiguration extends BaseConsumerConfiguration {// 省略其他代码@Beanpublic MmcKafkaInputerContainer mmcKafkaInputerContainer(MmcKafkaProcessorFactory factory,MmcKafkaBeanPostProcessor beanPostProcessor) throws Exception {// 省略其他代码// 逐个遍历，并生成consumerfor (Map.Entry<String, MmcMultiKafkaProperties.MmcKafkaProperties> entry : kafkas.entrySet()) {// 省略其他代码// 是否开启if (properties.isEnabled() && CommonUtil.isNotBlank(properties.getGroupId())) {// 省略其他代码// 设置容器inputer.setContainer(container);inputer.setName(name);inputer.setProperties(properties);inputer.init(); // 增加初始化// 省略其他代码}}return new MmcKafkaInputerContainer(inputers);}
}

4、由于增加了inputer增加了init方法，所以超级父类KafkaAbstractProcessor也增加一个默认实现。

@Slf4j
@Setter
public abstract class KafkaAbstractProcessor<T> implements MmcInputer {// 省略其他代码@Overridepublic void init() {}
}

5、新增MmcKafkaParallelAbstractProcessor并发处理类，根据多线程并发框架mmc-juc的特性，配置初始化多线程任务容器，并保留很多回调函数，方便子类覆盖重写。


@Slf4j
@Setter
public abstract class MmcKafkaParallelAbstractProcessor<T, R> extends MmcKafkaAbstractProcessor<T> {/*** taskExecutor.*/protected MmcTaskExecutor<T, R> taskExecutor;/*** init.*/public void init() {ContainerConfig config = properties.getContainer();this.taskExecutor = MmcTaskExecutor.<T, R>builder().taskProcessor(this::handelBatchDatas).threshold(config.getThreshold()).rateLimiter(buildRateLimiter(config.getRate())).taskMerger(this::mergeResult).forkJoinPoolConcurrency(config.getParallelism()).build();}@Overrideprotected void dealMessage(List<T> datas) throws ExecutionException, InterruptedException {if (properties.getContainer().isEnabled()) {// 开启并发处理R result = taskExecutor.execute(MmcTask.<T, R>builder().taskSource(datas).taskName(getTaskName(datas)).build());dealMessageCallBack(result);} else {// 同步处理R result = handelBatchDatas(datas);dealMessageCallBack(result);}}/*** 合并小任务结果（默认不合并）.** @param left 左边处理结果* @param right 右边处理结果* @return 合并后的结果*/protected R mergeResult(R left, R right) {return null;}/*** 构建速率限制器.** @param rate qps* @return 速率限制器*/protected RateLimiter buildRateLimiter(int rate) {return new TokenBucket(rate, rate);}/*** 当所有消息处理完后，会调用该方法.** @param result 处理结果*/protected void dealMessageCallBack(R result) {// default null}/*** 获取任务名称.*/protected String getTaskName(List<T> datas) {return name;}/*** 真正处理消息的方法.** @param datas 待处理消息* @return 小任务处理完的结果*/protected abstract R handelBatchDatas(List<T> datas);}

五、测试一下

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

       <dependency><groupId>io.github.vipjoey</groupId><artifactId>mmc-juc</artifactId><version>1.1</version></dependency>

2、增加并发消费者配置，生产者配置不变。

## json消息消费者
spring.kafka.five.enabled=true
spring.kafka.five.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.five.topic=mmc-topic-five
spring.kafka.five.group-id=group-consumer-five
spring.kafka.five.processor=fiveProcessor
spring.kafka.five.duplicate=true
spring.kafka.five.snakeCase=false
spring.kafka.five.consumer.auto-offset-reset=latest
spring.kafka.five.consumer.max-poll-records=10
spring.kafka.five.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.five.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
## 并发配置
spring.kafka.five.container.threshold=2
spring.kafka.five.container.rate=1000
spring.kafka.five.container.parallelism=8## json消息生产者
spring.kafka.five.enabled=true
spring.kafka.five.producer.name=fiveKafkaSender
spring.kafka.five.producer.bootstrap-servers=${spring.embedded.kafka.brokers}
spring.kafka.five.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.five.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3、编写测试类。


@Slf4j
@Service("fiveProcessor")
public class FiveProcessor extends MmcKafkaParallelAbstractProcessor<ParalleMsg, Void> {@Overrideprotected Void handelBatchDatas(List<ParalleMsg> datas) {datas.forEach(x -> {log.info("handelBatchDatas one: {}", x);});return null;}
}@Slf4j
@ActiveProfiles("dev")
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = {MmcMultiProducerAutoConfiguration.class, MmcMultiConsumerAutoConfiguration.class,FiveProcessor.class})
@TestPropertySource(value = "classpath:application-paralle.properties")
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"},topics = {"${spring.kafka.five.topic}"})
public class KafkaParalleMessageTest {@Value("${spring.kafka.five.topic}")private String fiveTopic;@Resource(name = "fiveKafkaSender")private MmcKafkaSender mmcKafkaSender;@Testvoid testDealMessage() throws Exception {Thread.sleep(2 * 1000);// 模拟生产数据produceMessage();Thread.sleep(10 * 1000);}void produceMessage() {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);mmcKafkaSender.sendStringMessage(fiveTopic, "aaa", json);}}
}

5、运行一下，测试通过，可以看到能正常发送消息和消费。
在这里插入图片描述

五、小结

将本项目代码构建成starter，就可以大大提升我们开发效率，我们只需要关心业务代码的开发，github项目源码：轻触这里。如果对你有用可以打个星星哦。

  • 《搭建大型分布式服务（三十六）SpringBoot 零代码方式整合多个kafka数据源》
  • 《搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符》
  • 《搭建大型分布式服务（三十八）SpringBoot 整合多个kafka数据源-支持protobuf》
  • 《搭建大型分布式服务（三十九）SpringBoot 整合多个kafka数据源-支持Aware模式》
  • 《搭建大型分布式服务（四十）SpringBoot 整合多个kafka数据源-支持生产者》
  • 《搭建大型分布式服务（四十一）SpringBoot 整合多个kafka数据源-支持亿级消息生产者》
  • 《搭建大型分布式服务（四十二）SpringBoot 无代码侵入实现多Kafka数据源整合插件发布》
  • 《搭建大型分布式服务（四十三）SpringBoot 多Kafka数据源发布到Maven中央仓库：让世界看到你的作品！》
  • 《搭建大型分布式服务（四十四）SpringBoot 无代码侵入实现多Kafka数据源：单分区提升至十万级消费速度！》

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

这篇关于搭建大型分布式服务（四十四）SpringBoot 无代码侵入实现多Kafka数据源：单分区提升至十万级消费速度！的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/1129883。

相关文章

Android实现悬浮按钮功能

Android实现悬浮按钮功能

《Android实现悬浮按钮功能》在很多场景中,我们希望在应用或系统任意界面上都能看到一个小的“悬浮按钮”（FloatingButton）,用来快速启动工具、展示未读信息或快捷操作,所以本文给大家介绍... 目录一、项目概述二、相关技术知识三、实现思路四、整合代码4.1 Java 代码（MainActivi
阅读更多...
Java数组初始化的五种方式

Java数组初始化的五种方式

《Java数组初始化的五种方式》数组是Java中最基础且常用的数据结构之一,其初始化方式多样且各具特点,本文详细讲解Java数组初始化的五种方式,分析其适用场景、优劣势对比及注意事项,帮助避免常见陷阱... 目录1. 静态初始化：简洁但固定代码示例核心特点适用场景注意事项2. 动态初始化：灵活但需手动管理代
阅读更多...
Java使用SLF4J记录不同级别日志的示例详解

Java使用SLF4J记录不同级别日志的示例详解

《Java使用SLF4J记录不同级别日志的示例详解》SLF4J是一个简单的日志门面,它允许在运行时选择不同的日志实现,这篇文章主要为大家详细介绍了如何使用SLF4J记录不同级别日志,感兴趣的可以了解下... 目录一、SLF4J简介二、添加依赖三、配置Logback四、记录不同级别的日志五、总结一、SLF4J
阅读更多...
将Java项目提交到云服务器的流程步骤

将Java项目提交到云服务器的流程步骤

《将Java项目提交到云服务器的流程步骤》所谓将项目提交到云服务器即将你的项目打成一个jar包然后提交到云服务器即可,因此我们需要准备服务器环境为：Linux+JDK+MariDB（MySQL）+Gi... 目录1. 安装 jdk1.1 查看 jdk 版本1.2 下载 jdk2. 安装 mariadb(my
阅读更多...
使用Python实现一个优雅的异步定时器

使用Python实现一个优雅的异步定时器

《使用Python实现一个优雅的异步定时器》在Python中实现定时器功能是一个常见需求,尤其是在需要周期性执行任务的场景下,本文给大家介绍了基于asyncio和threading模块,可扩展的异步定... 目录需求背景代码1. 单例事件循环的实现2. 事件循环的运行与关闭3. 定时器核心逻辑4. 启动与停
阅读更多...
基于Python实现读取嵌套压缩包下文件的方法

基于Python实现读取嵌套压缩包下文件的方法

《基于Python实现读取嵌套压缩包下文件的方法》工作中遇到的问题,需要用Python实现嵌套压缩包下文件读取,本文给大家介绍了详细的解决方法,并有相关的代码示例供大家参考,需要的朋友可以参考下... 目录思路完整代码代码优化思路打开外层zip压缩包并遍历文件：使用with zipfile.ZipFil
阅读更多...
Python实现word文档内容智能提取以及合成

Python实现word文档内容智能提取以及合成

《Python实现word文档内容智能提取以及合成》这篇文章主要为大家详细介绍了如何使用Python实现从10个左右的docx文档中抽取内容,再调整语言风格后生成新的文档,感兴趣的小伙伴可以了解一下... 目录核心思路技术路径实现步骤阶段一：准备工作阶段二：内容提取 (python 脚本)阶段三：语言风格调
阅读更多...
SpringBoot中配置Redis连接池的完整指南

SpringBoot中配置Redis连接池的完整指南

《SpringBoot中配置Redis连接池的完整指南》这篇文章主要为大家详细介绍了SpringBoot中配置Redis连接池的完整指南,文中的示例代码讲解详细,具有一定的借鉴价值,感兴趣的小伙伴可以... 目录一、添加依赖二、配置 Redis 连接池三、测试 Redis 操作四、完整示例代码（一）pom.
阅读更多...
Java 正则表达式URL 匹配与源码全解析

Java 正则表达式URL 匹配与源码全解析

《Java正则表达式URL匹配与源码全解析》在Web应用开发中,我们经常需要对URL进行格式验证,今天我们结合Java的Pattern和Matcher类,深入理解正则表达式在实际应用中... 目录1.正则表达式分解：2. 添加域名匹配 (2)3. 添加路径和查询参数匹配 (3) 4. 最终优化版本5.设计思
阅读更多...
C#实现将Excel表格转换为图片(JPG/ PNG)

C#实现将Excel表格转换为图片(JPG/ PNG)

《C#实现将Excel表格转换为图片(JPG/PNG)》Excel表格可能会因为不同设备或字体缺失等问题,导致格式错乱或数据显示异常,转换为图片后,能确保数据的排版等保持一致,下面我们看看如何使用C... 目录通过C# 转换Excel工作表到图片通过C# 转换指定单元格区域到图片知识扩展C# 将 Excel
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2