系列文章目录
文章目录
- 系列文章目录
- 前言
- 一、本文要点
- 二、开发环境
- 三、原项目
- 四、修改项目
- 五、测试一下
- 五、小结
前言
本插件稳定运行上百个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.StringDeserializer4、编写测试类，测试类保持不变。
@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模式》
加我加群一起交流学习！更多干货下载、项目源码和大厂内推等着你
 |  |
![LeetCode - 贪心(Greedy)算法集合(Python)[分配问题|区间问题]](https://img-blog.csdnimg.cn/direct/b023f7cbce564612830e58c7a2f174d2.png)
