当前位置: 首页 > news >正文

搭建大型分布式服务(四十二)SpringBoot 无代码侵入实现多Kafka数据源整合插件发布

系列文章目录


文章目录

  • 系列文章目录
  • 前言
  • MultiKafkaStarter [V2.2]
    • 一、功能特性
    • 二、快速开始(生产端)
    • 三、快速开始(消费端)
    • 四、其它特性
    • 五、变更记录
    • 六、参考文章


前言

在分布式服务的架构演进中,消息队列作为核心组件之一,承载着解耦、异步、削峰填谷等关键职责。Apache Kafka 作为业界广泛使用的分布式流处理平台,因其高吞吐、低延迟的特性被大量应用在各类大数据场景中。然而,随着业务的复杂度不断提升,如何在 SpringBoot 中高效地整合并管理多个 Kafka 数据源,成为了一个值得探讨的问题。

在过去的一段时间里,我们通过系列文章详细阐述了如何在 SpringBoot 中以零代码或极低的代码侵入方式,实现多 Kafka 数据源的整合。从基础的配置到高级特性如 protobuf 支持、Aware 模式以及亿级消息生产者的优化,我们希望通过这些内容帮助开发者更加便捷地应对复杂的业务场景。

今天,我们将这些内容凝练成一个全新的 SpringBoot 插件——MultiKafkaStarter,旨在进一步降低开发者整合多 Kafka 数据源的门槛,提升系统的可维护性和扩展性。

核心特点

  • 无代码侵入:通过 SpringBoot 的自动配置机制,无需修改业务代码即可实现多 Kafka 数据源的整合。
  • 灵活配置:支持动态配置多个 Kafka 数据源,包括 bootstrap servers、group id、security protocol 等关键参数。
  • 全面特性支持:不仅支持基础的消息消费和生产功能,还提供了对 protobuf 序列化/反序列化的支持,以及对 Aware 模式的适配。
  • 亿级消息处理:针对高并发场景,提供了包括批量发送、分区策略优化等在内的多项性能优化措施,确保系统能够稳定处理亿级消息量。
  • 易用性与可维护性:插件采用模块化的设计思想,易于集成和升级,同时提供了丰富的文档和社区支持

国籍惯例,先上源码:Github源码

MultiKafkaStarter [V2.2]

SpringBoot 零代码方式整合多个kafka数据源,支持任意kafka集群,已封装为一个小模块,集成所有kafka配置,让注意力重新回归业务本身。

一、功能特性

  • SpringBoot无编程方式整合多个kafka数据源
  • 支持批量消费kafka并对单批次消息根据条件去重
  • 支持消费kafka消息类型为pb格式
  • 支持任意数量生产者

1、引入最新依赖包,如果找不到依赖包,请到工程目录mvn clean package install执行一下命令。

<dependency><groupId>io.github.vipjoey</groupId><artifactId>multi-kafka-starter</artifactId><version>2.2</version>
</dependency>

二、快速开始(生产端)

2、添加kafka地址等相关配置。


## json消息生产者
spring.kafka.four.enabled=true
spring.kafka.four.producer.count=1 ## 生产者数量,默认为1个
spring.kafka.four.producer.name=fourKafkaSender  ## 设置bean的名称,方便后续引用。如果没有设置,默认值为xxxKafkaSender
spring.kafka.four.producer.bootstrap-servers=${spring.embedded.kafka.brokers} ## 必须设置
spring.kafka.four.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.four.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer## pb 消息生产者
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.ByteArraySerializer

3、根据名称注入生产者MmcKafkaMultiSender,就可以发送kafka消息。

@Resource(name = "fourKafkaSender")private MmcKafkaMultiSender mmcKafkaMultiSender;@Resource(name = "fiveKafkaSender")private MmcKafkaMultiSender mmcKafkaMultiSender;@Resourceprivate MmcKafkaOutputContainer mmcKafkaOutputContainer;// 方式一void produceMessage() {for (int i = 0; i < 10; i++) {DemoAwareMsg msg = new DemoAwareMsg();msg.setRoutekey("routekey" + i);msg.setName("name" + i);msg.setTimestamp(System.currentTimeMillis());String json = JsonUtil.toJsonStr(msg);mmcKafkaMultiSender.sendStringMessage(topicOne, "aaa", json);}}// 方式二void produceMessage() {MmcKafkaSender sender = mmcKafkaOutputContainer.getOutputs().get("xxxKafkaSender");sender.sendStringMessage(topic, sku.getRoutekey(), message);}

三、快速开始(消费端)

2、添加kafka地址等相关配置。

## 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=你的处理类bean名称(例如:oneProcessor)
spring.kafka.one.dupicate=true   ## 如果为true表示对批次内的kafka消息去重,需要实现MmcKafkaMsg接口,默认为false
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=你的处理类bean名称
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## protobuf类型的消息的kafka配置
spring.kafka.pb.enabled=true
spring.kafka.pb.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.pb.topic=mmc-topic-pb
spring.kafka.pb.group-id=group-consumer-pb
spring.kafka.pb.processor=pbProcessor
spring.kafka.pb.consumer.auto-offset-reset=latest
spring.kafka.pb.consumer.max-poll-records=10
spring.kafka.pb.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.pb.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer

3、新建kafka消息对应的实体类,可以选择实现MmcMsgDistinctAware接口,例如

@Data
class DemoMsg implements MmcMsgDistinctAware {private String routekey;private String name;private Long timestamp;}如果你配置了spring.kafka.xxx.duplicate=fale,则不需要实现MmcMsgDistinctAware接口。PS:如果实现MmcMsgDistinctAware接口,就自动具备了消息去重能力

4、新建kafka消息处理类,要求继承MmcKafkaKafkaAbastrctProcessor,然后就可以愉快地编写你的业务逻辑了。

@Slf4j
@Service("oneProcessor") // 你的处理类bean名称,如果没有定义bean名称,那么默认就是首字母缩写的类名称
public class OneProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {@Overrideprotected void dealMessage(List<DemoMsg> datas) {// 下面开始编写你的业务代码}}@Slf4j
@Service("pbProcessor")
public class PbProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {@Overrideprotected Stream<DemoMsg> doParseProtobuf(byte[] record) {try {DemoPb.PbMsg msg = DemoPb.PbMsg.parseFrom(record);DemoMsg demo = new DemoMsg();BeanUtils.copyProperties(msg, demo);return Stream.of(demo);} catch (InvalidProtocolBufferException e) {log.error("parssPbError", e);return Stream.empty();}}@Overrideprotected void dealMessage(List<DemoMsg> datas) {System.out.println("PBdatas: " + datas);}
}

四、其它特性

1、支持单次拉取kafka的batch消息里去重,需要实现MmcMsgDistinctAware的getRoutekey()和getTimestamp()方法;如果为false,则不要实现MmcMsgDistinctAware接口。

spring.kafka.xxx.duplicate=true

2、支持字符串kafka消息,json是驼峰或者下划线

# 默认为支持驼峰的kafka消息,为ture则支持下划线的消息
spring.kafka.xxx.snakeCase=false

3、支持pb的kafka消息,需要自行重写父类的doParseProtobuf方法;

    @Overrideprotected Stream<DemoMsg> doParseProtobuf(byte[] record) {try {DemoMsg msg = new DemoMsg();DemoPb.PbMsg pb = DemoPb.PbMsg.parseFrom(record);BeanUtils.copyProperties(pb, msg);return Stream.of(msg);} catch (InvalidProtocolBufferException e) {log.error("doParseProtobuf error: {}", e.getMessage());return Stream.empty();}}

4、支持获取kafka的topic、offset属性,注入到实体类中,需要实现MmcMsgKafkaAware接口

@Data
class DemoAwareMsg implements MmcKafkaAware {private String routekey;private String name;private Long timestamp;private String topic;private long offset;}

五、变更记录

  • 20240623 v2.2 支持Kafka生产者,并对MultiKafkaConsumerStarter项目重命名为MultiKafkaStarter
  • 20240602 v2.1 支持获取kafka消息中topic、offset属性
  • 20240602 v2.0 支持protobuf、json格式
  • 20240430 v1.1 取消限定符
  • 20231111 v1.0 初始化

六、参考文章

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

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

相关文章:

搭建大型分布式服务(四十二)SpringBoot 无代码侵入实现多Kafka数据源整合插件发布

系列文章目录 文章目录 系列文章目录前言MultiKafkaStarter [V2.2]一、功能特性二、快速开始&#xff08;生产端&#xff09;三、快速开始&#xff08;消费端&#xff09;四、其它特性五、变更记录六、参考文章 前言 在分布式服务的架构演进中&#xff0c;消息队列作为核心组件…...

Python 学习路线及技巧

一、学习路线 1. 基础阶段 ● 学习 Python 的语法基础&#xff0c;如变量、数据类型、运算符、控制流等。 ● 掌握常用的 Python 标准库&#xff0c;如 os、sys、re、datetime 等。 ● 通过编写简单的程序来巩固基础&#xff0c;如计算器、字符串处理等。 2. 进阶阶段 ● 深入…...

计算机网络知识整理笔记

目录 1.对网络协议的分层&#xff1f; 2.TCP/IP和UDP之间的区别&#xff1f; 3.建立TCP连接的三次握手&#xff1f; 4.断开TCP连接的四次挥手&#xff1f; 5.TCP协议如何保证可靠性传输&#xff1f; 6.什么是TCP的拥塞控制&#xff1f; 7.什么是HTTP协议&#xff1f; 8…...

练习 String翻转 注册处理 字符串统计

p493 将字符串中指定部分进行翻转 package chapter;public class reverse {public static void main(String[] args) {String str "abcdef";str reverseMethod(str,0,3);System.out.println(str);}public static String reverseMethod(String str, int start, in…...

linux的常用系统维护命令

1.ps显示某个时间点的程序运行情况 -a &#xff1a;显示所有用户的进程 -u &#xff1a;显示用户名和启动时间 -x &#xff1a;显示 没有控制终端的进程 -e &#xff1a;显示所有进程&#xff0c;包括没有控制终端的进程 -l &#xff1a;长格式显示 -w &#xff1a;宽…...

java:aocache 0.4.0 缓存控制机制

aoocache发布第一个版本0.1.0时&#xff0c;没有考虑到使用aocache的项目对方法缓存的控制需求。 场景 给同事做培训时&#xff0c;同事提到这个需求&#xff0c;他希望能够有方法主动去清理指定方法的缓存&#xff1a; 他的数据是由其他服务启动时提供的&#xff0c;他的方法…...

试析C#编程语言的特点及功能

行步骤&#xff0c;而不必创建新方法。其声明方法是在实例化委托基础上&#xff0c;加一对花括号以代表执行范围&#xff0c;再加一个分号终止语句。 2.3.3 工作原理 C#编译器在“匿名”委托时会自动把执行代码转换成惟一命名类里的惟一命名函数。再对存储代码块的委托进行设…...

Textual Learning2 -- 使用时的小问题

1、出现的问题&#xff1a; 在vscode里面直接运行函数会显示报错&#xff1a; 我尝试在vscode中含textual库的环境下运行&#xff0c;但仍然报错 2、解决方案&#xff1a; 在命令行中运行&#xff1a; 首先按winR&#xff0c;输入cmd打开命令行 或在已经安装的conda环境&a…...

CST--如何在PCB三维模型中自由创建离散端口

在使用CST电磁仿真软件进行PCB的三维建模时&#xff0c;经常会遇到不能自动创建离散端口的问题&#xff0c;原因有很多&#xff0c;比如&#xff1a;缺少元器件封装、开路端口、多端子模型等等&#xff0c;这个时候&#xff0c;很多人会选择手动进行端口创建&#xff0c;但是&a…...

C++中的虚函数表结构框架

一.虚函数表介绍 Virtual Table虚函数表是实现多态的 每个有虚函数的类的实现&#xff0c;都有个指向虚函数的指针表&#xff08;不管是父类还是子类&#xff09; 指向虚表的指针是作为数据成员存在实例对象中 当调用虚函数时&#xff0c;就去查找对象的虚表中指向整顿派生类函…...

【ES】--Elasticsearch的高亮模式

目录 一、高亮策略1、Fast Vector Highlighter(快速向量高亮器)2、Posting Highlighter(帖子高亮器)3、Unified Highlighter(统一高亮器)4、Plain Highlighter(普通高亮器)5、总结二、高亮参数三、高亮案例解析1、words_one配置解析2、words_two配置解析3、words_three…...

使用matlab开发stm32总结,stm32-matlab常见的问题处理以及报错合集

1&#xff0c;问题&#xff1a;本来是好的&#xff0c;突然编译运行报错&#xff0c;说是确少包&#xff0c; 解决方案&#xff1a;重启以后好了 2&#xff0c;有完美的马鞍波&#xff0c;为什么不能够转动呢&#xff1f; 原因是我这里模型的问题&#xff0c;我计算出来的是占…...

落石滑坡监测报警系统:创新保障高速公路安全

​ ​​在现代交通建设中&#xff0c;高速公路的安全性和稳定性至关重要。特别是易发生落石区域&#xff0c;如何有效预防和应对落石滑坡带来的事故成为了一项关键性挑战。为此&#xff0c;落石滑坡监测报警系统应运而生&#xff0c;它通过先进的技术手段&#xff0c;为高速…...

Linux开发讲课20--- QSPI

SPI 是英语 Serial Peripheral interface 的缩写&#xff0c;顾名思义就是串行外围设备接口&#xff0c;一种高速的&#xff0c;全双工&#xff0c;同步的通信总线&#xff0c;并且在芯片的管脚上只占用四根线&#xff0c;节约了芯片的管脚&#xff0c;为 PCB 的布局上节省空间…...

VMware ESXi 8.0U3 macOS Unlocker OEM BIOS 集成驱动版,新增 12 款 I219 网卡驱动

VMware ESXi 8.0U3 macOS Unlocker & OEM BIOS 集成驱动版&#xff0c;新增 12 款 I219 网卡驱动 VMware ESXi 8.0U3 macOS Unlocker & OEM BIOS 集成网卡驱动和 NVMe 驱动 (集成驱动版) 发布 ESXi 8.0U3 集成驱动版&#xff0c;在个人电脑上运行企业级工作负载 请访…...

vuepress使用简介及个人博客搭建

目录 一、介绍二、环境准备三、安装运行vuepress四、目录结构五、配置文件六、导航栏配置七、导航栏logo八、浏览器图标九、侧边栏配置十、添加 Git 仓库和编辑链接十一、部署到GitHub十二、搭建成功 一、介绍 VuePress 是 Vuejs 官方提供的一个是Vue驱动的静态网站生成器&…...

c#文件读写

1.1读取文件 方法说明​File.ReadAllText(FilePath);​读取指定路径的文件​File.ReadAllText(FilePath, Encoding);​通过指定编码格式来读取指定文件​File.ReadAllBytes();​读取二进制文件&#xff0c;并把内容读取到一个字节数组​File.ReadAllLines();​以行的形式读取文…...

WIFI 企业级认证手段 EAP-TLS介绍

EAP-TLS&#xff08;EAP-Transport Layer Security&#xff09;被认为是WLAN网络里最安全的认证方法&#xff0c;因此被企业广泛采用。本文会针对EAP-TLS的基本原理进行介绍。 在介绍原理之前&#xff0c;先介绍下WLAN网络里认证加密手段涉及到的一些基本概念。 1 802.1x IEE…...

【网络架构】keepalive

目录 一、keepalive基础 1.1 作用 1.2 原理 1.3 功能 二、keepalive安装 2.1 yum安装 2.2 编译安装 三、配置文件 3.1 keepalived相关文件 3.2 主配置的组成 3.2.1 全局配置 3.2.2 配置虚拟路由器 四、实际操作 4.1 lvskeepalived高可用群集 4.2 keepalivedngi…...

【Dison夏令营 Day 03】使用 Python 创建我们自己的 21 点游戏

21 点(英文&#xff1a;Blackjack)是一种在赌场玩的纸牌游戏。这种游戏的参与者不是互相竞争&#xff0c;而是与赌场指定的庄家竞争。在本文中&#xff0c;我们将从头开始创建可在终端上玩的玩家与庄家之间的二十一点游戏。 二十一点规则 我们将为从未玩过二十一点的读者提供…...

多云管理“拦路虎”:深入解析网络互联、身份同步与成本可视化的技术复杂度​

一、引言&#xff1a;多云环境的技术复杂性本质​​ 企业采用多云策略已从技术选型升维至生存刚需。当业务系统分散部署在多个云平台时&#xff0c;​​基础设施的技术债呈现指数级积累​​。网络连接、身份认证、成本管理这三大核心挑战相互嵌套&#xff1a;跨云网络构建数据…...

云计算——弹性云计算器(ECS)

弹性云服务器&#xff1a;ECS 概述 云计算重构了ICT系统&#xff0c;云计算平台厂商推出使得厂家能够主要关注应用管理而非平台管理的云平台&#xff0c;包含如下主要概念。 ECS&#xff08;Elastic Cloud Server&#xff09;&#xff1a;即弹性云服务器&#xff0c;是云计算…...

爬虫基础学习day2

# 爬虫设计领域 工商&#xff1a;企查查、天眼查短视频&#xff1a;抖音、快手、西瓜 ---> 飞瓜电商&#xff1a;京东、淘宝、聚美优品、亚马逊 ---> 分析店铺经营决策标题、排名航空&#xff1a;抓取所有航空公司价格 ---> 去哪儿自媒体&#xff1a;采集自媒体数据进…...

【开发技术】.Net使用FFmpeg视频特定帧上绘制内容

目录 一、目的 二、解决方案 2.1 什么是FFmpeg 2.2 FFmpeg主要功能 2.3 使用Xabe.FFmpeg调用FFmpeg功能 2.4 使用 FFmpeg 的 drawbox 滤镜来绘制 ROI 三、总结 一、目的 当前市场上有很多目标检测智能识别的相关算法&#xff0c;当前调用一个医疗行业的AI识别算法后返回…...

什么是Ansible Jinja2

理解 Ansible Jinja2 模板 Ansible 是一款功能强大的开源自动化工具&#xff0c;可让您无缝地管理和配置系统。Ansible 的一大亮点是它使用 Jinja2 模板&#xff0c;允许您根据变量数据动态生成文件、配置设置和脚本。本文将向您介绍 Ansible 中的 Jinja2 模板&#xff0c;并通…...

Hive 存储格式深度解析:从 TextFile 到 ORC,如何选对数据存储方案?

在大数据处理领域&#xff0c;Hive 作为 Hadoop 生态中重要的数据仓库工具&#xff0c;其存储格式的选择直接影响数据存储成本、查询效率和计算资源消耗。面对 TextFile、SequenceFile、Parquet、RCFile、ORC 等多种存储格式&#xff0c;很多开发者常常陷入选择困境。本文将从底…...

基于Java Swing的电子通讯录设计与实现:附系统托盘功能代码详解

JAVASQL电子通讯录带系统托盘 一、系统概述 本电子通讯录系统采用Java Swing开发桌面应用&#xff0c;结合SQLite数据库实现联系人管理功能&#xff0c;并集成系统托盘功能提升用户体验。系统支持联系人的增删改查、分组管理、搜索过滤等功能&#xff0c;同时可以最小化到系统…...

无人机侦测与反制技术的进展与应用

国家电网无人机侦测与反制技术的进展与应用 引言 随着无人机&#xff08;无人驾驶飞行器&#xff0c;UAV&#xff09;技术的快速发展&#xff0c;其在商业、娱乐和军事领域的广泛应用带来了新的安全挑战。特别是对于关键基础设施如电力系统&#xff0c;无人机的“黑飞”&…...

08. C#入门系列【类的基本概念】:开启编程世界的奇妙冒险

C#入门系列【类的基本概念】&#xff1a;开启编程世界的奇妙冒险 嘿&#xff0c;各位编程小白探险家&#xff01;欢迎来到 C# 的奇幻大陆&#xff01;今天咱们要深入探索这片大陆上至关重要的 “建筑”—— 类&#xff01;别害怕&#xff0c;跟着我&#xff0c;保准让你轻松搞…...

【C++进阶篇】智能指针

C内存管理终极指南&#xff1a;智能指针从入门到源码剖析 一. 智能指针1.1 auto_ptr1.2 unique_ptr1.3 shared_ptr1.4 make_shared 二. 原理三. shared_ptr循环引用问题三. 线程安全问题四. 内存泄漏4.1 什么是内存泄漏4.2 危害4.3 避免内存泄漏 五. 最后 一. 智能指针 智能指…...