搭建大型分布式服务（三十九）SpringBoot 整合多个kafka数据源-支持Aware模式

2024-06-02 18:28
文章标签 java 模式 springboot spring 搭建 整合 支持 多个 数据源 kafka 分布式服务 大型 三十九 aware

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

系列文章目录

文章目录

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

前言

本插件稳定运行上百个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## pb 消息消费者
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

国籍惯例，先上源码：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等参数

二、开发环境

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

三、原项目

1、接前文，我们修改了抽象父类，并下沉了kafka相关依赖，使得可以支持消费pb类型的格式数据，根据自己的需求解析出实体类。但也有小伙伴反馈，有时候需要获取kafka消息中的partition、offset、topic、header等参数，这个怎样办？


@Slf4j
@Service
public class OneProcessor extends MmcKafkaKafkaAbastrctProcessor<DemoMsg> {// 需要在DemoMsg获取partition、offset、topic、header等参数// 但不又不想重写父类的doParse()方法，这样显得太笨重// 因为需要获取kafka消息中的参数毕竟场景不算太多@Overrideprotected void dealMessage(List<DemoMsg> datas) {datas.forEach(x -> {log.info("dealMessage one: {}", x);});}}

2、在Spring Boot中，Aware接口是一种特殊的接口，它允许我们在Spring容器初始化时获取特定的bean或属性。Spring Boot提供了多种Aware接口，用于获取不同类型的bean或属性。这些接口的实现原理主要依赖于Spring框架的生命周期回调机制。所以，我们是否可以参考这种方式，让子类按需选择需要哪些参数能力，类似下面这样呢？

// 获取消息去重能力
@Data
class DemoMsg implements MmcMsgDistinctAware{private String routekey;private String name;private Long timestamp;}// 获取kafka消息的topic、offset参数
@Data
class DemoMsg implements MmcMsgKafkaAware {private String routekey;private String name;private Long timestamp;private String topic;private long offset;
}

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

四、修改项目

1、重命名MmcKafkaMsg类为MmcMsgDistinctAware，使得更加符合规范，只要子类实体类实现了本接口，那么就可以具备消息去重的能力。

public interface MmcMsgDistinctAware {/*** 代表kafka消息的唯一键，用于批次内分组.* @return 唯一键*/String getRoutekey();/*** kafka消息生产或接收时间，用于批次内分组，根据时间去重，取最新的消息.** @return 消息时间*/Long getTimestamp();
}

2、新增MmcMsgKafkaAware，只要子类的实体类实现本接口，就可以方便获取kafka消息中的topic、offset等参数。

public interface MmcMsgKafkaAware {/*** 注入topic.** @param topic topic名称*/void setTopic(String topic);/*** 注入offset.** @param offset offset*/void setOffset(long offset);
}

3、修改KafkaAbastrctProcessor抽象父类，重写解析消息方法，使得可以根据实体类的Aware接口标记来获取对应的能力；

@Slf4j
@Setter
abstract class KafkaAbstractProcessor<T> implements MmcInputer {// 重写解析消息方法，使得可以根据实体类的Aware接口标记来获取对应的能力/*** 将kafka消息解析为实体，支持json对象或者json数组.** @param map kafka消息对象，包含key、value、topic、partition、offset等* @return 实体类*/protected Stream<T> doParse(ConsumerRecord<String, Object> map) {// 消息对象Object record = map.value();// 如果是pb格式if (record instanceof byte[]) {return doParseProtobuf((byte[]) record);} else if (record instanceof String) {// 普通kafka消息String json = record.toString();if (json.startsWith("[")) {// 数组List<T> datas = doParseJsonArray(json);if (CommonUtil.isEmpty(datas)) {log.warn("{} doParse error, json={} is error.", name, json);return Stream.empty();}// 反序列对象后，做一些初始化操作datas = datas.stream().peek(x -> doKafkaAware(x, map)).peek(this::doAfterParse).collect(Collectors.toList());return datas.stream();} else {// 对象T data = doParseJsonObject(json);if (null == data) {log.warn("{} doParse error, json={} is error.", name, json);return Stream.empty();}// 注入kafka相关doKafkaAware(data, map);// 反序列对象后，做一些初始化操作doAfterParse(data);return Stream.of(data);}} else if (record instanceof MmcKafkaMsg) {// 如果本身就是MmcKafkaMsg对象，直接返回//noinspection uncheckedreturn Stream.of((T) record);} else {throw new UnsupportedForMessageFormatException("not support message type");}}protected void doKafkaAware(T x, ConsumerRecord<String, Object> record) {// 根据自己诉求去扩展，可以增加无限xxxAware，获取任意record的参数if (x instanceof MmcMsgKafkaAware) {((MmcMsgKafkaAware) x).setOffset(record.offset());((MmcMsgKafkaAware) x).setTopic(record.topic());}}

五、测试一下

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><dependency><groupId>com.google.protobuf</groupId><artifactId>protobuf-java</artifactId><version>3.18.0</version><scope>test</scope></dependency><dependency><groupId>com.google.protobuf</groupId><artifactId>protobuf-java-util</artifactId><version>3.18.0</version><scope>test</scope></dependency>

2、修改DemoMsg，让它实现MmcMsgKafkaAware接口，用来获取topic和offset参数，Processor不用修改，保持不变；

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

3、消费者配置保持不变。

## 消费者配置
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 {Thread.sleep(2 * 1000);// 模拟生产数据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、运行一下，测试通过，可以看到已经打印topic和offset。
在这里插入图片描述

五、小结

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

《搭建大型分布式服务（三十六）SpringBoot 零代码方式整合多个kafka数据源》
《搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符》
《搭建大型分布式服务（三十八）SpringBoot 整合多个kafka数据源-支持protobuf》
《搭建大型分布式服务（三十九）SpringBoot 整合多个kafka数据源-支持Aware模式》

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

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


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

相关文章

Java五子棋之坐标校正

Java五子棋之坐标校正

上篇针对了Java项目中的解构思维，在这篇内容中我们不妨从整体项目中拆解拿出一个非常重要的五子棋逻辑实现：坐标校正，我们如何使漫无目的鼠标点击变得有序化和可控化呢？ 目录 一、从鼠标监听到获取坐标 1.MouseListener和MouseAdapter 2.mousePressed方法 二、坐标校正的具体实现方法 1.关于fillOval方法 2.坐标获取 3.坐标转换 4.坐
阅读更多...
Spring Cloud：构建分布式系统的利器

Spring Cloud：构建分布式系统的利器

引言 在当今的云计算和微服务架构时代，构建高效、可靠的分布式系统成为软件开发的重要任务。Spring Cloud 提供了一套完整的解决方案，帮助开发者快速构建分布式系统中的一些常见模式（例如配置管理、服务发现、断路器等）。本文将探讨 Spring Cloud 的定义、核心组件、应用场景以及未来的发展趋势。 什么是 Spring Cloud Spring Cloud 是一个基于 Spring
阅读更多...
Javascript高级程序设计(第四版)--学习记录之变量、内存

Javascript高级程序设计(第四版)--学习记录之变量、内存

原始值与引用值 原始值：简单的数据即基础数据类型，按值访问。 引用值：由多个值构成的对象即复杂数据类型，按引用访问。 动态属性 对于引用值而言，可以随时添加、修改和删除其属性和方法。 let person = new Object();person.name = 'Jason';person.age = 42;console.log(person.name,person.age);//'J
阅读更多...
java8的新特性之一（Java Lambda表达式）

java8的新特性之一（Java Lambda表达式）

1：Java8的新特性 Lambda 表达式： 允许以更简洁的方式表示匿名函数（或称为闭包）。可以将Lambda表达式作为参数传递给方法或赋值给函数式接口类型的变量。 Stream API： 提供了一种处理集合数据的流式处理方式，支持函数式编程风格。 允许以声明性方式处理数据集合（如List、Set等）。提供了一系列操作，如map、filter、reduce等，以支持复杂的查询和转
阅读更多...
一份LLM资源清单围观技术大佬的日常；手把手教你在美国搭建「百万卡」AI数据中心；为啥大模型做不好简单的数学计算？ | ShowMeAI日报

一份LLM资源清单围观技术大佬的日常；手把手教你在美国搭建「百万卡」AI数据中心；为啥大模型做不好简单的数学计算？ | ShowMeAI日报

👀日报&周刊合集 | 🎡ShowMeAI官网 | 🧡 点赞关注评论拜托啦！ 1. 为啥大模型做不好简单的数学计算？从大模型高考数学成绩不及格说起 司南评测体系 OpenCompass 选取 7 个大模型 (6 个开源模型+ GPT-4o)，组织参与了 2024 年高考「新课标I卷」的语文、数学、英语考试，然后由经验丰富的判卷老师评判得分。 结果如上图所
阅读更多...
如何开启和关闭3GB模式

如何开启和关闭3GB模式

https://jingyan.baidu.com/article/4d58d5414dfc2f9dd4e9c082.html
阅读更多...
Java面试八股之怎么通过Java程序判断JVM是32位还是64位

Java面试八股之怎么通过Java程序判断JVM是32位还是64位

怎么通过Java程序判断JVM是32位还是64位 可以通过Java程序内部检查系统属性来判断当前运行的JVM是32位还是64位。以下是一个简单的方法： public class JvmBitCheck {public static void main(String[] args) {String arch = System.getProperty("os.arch");String dataM
阅读更多...
详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

详细分析Springmvc中的@ModelAttribute基本知识（附Demo）

目录 前言1. 注解用法1.1 方法参数1.2 方法1.3 类 2. 注解场景2.1 表单参数2.2 AJAX请求2.3 文件上传 3. 实战4. 总结 前言 将请求参数绑定到模型对象上，或者在请求处理之前添加模型属性 可以在方法参数、方法或者类上使用 一般适用这几种场景： 表单处理：通过 @ModelAttribute 将表单数据绑定到模型对象上预处理逻辑：在请求处理之前
阅读更多...
eclipse运行springboot项目，找不到主类

eclipse运行springboot项目，找不到主类

解决办法尝试了很多种，下载sts压缩包行不通。最后解决办法如图： help--->Eclipse Marketplace--->Popular--->找到Spring Tools 3---->Installed。
阅读更多...
JAVA读取MongoDB中的二进制图片并显示在页面上

JAVA读取MongoDB中的二进制图片并显示在页面上

1：Jsp页面： <td><img src="${ctx}/mongoImg/show"></td> 2：xml配置： <?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2