搭建大型分布式服务(四十)SpringBoot 整合多个kafka数据源-支持生产者

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

系列文章目录


文章目录

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


前言

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

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

例如下面这样简单的配置就完成SpringBoot和kafka的整合,我们只需要关心com.mmc.multi.kafka.starter.OneProcessorcom.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## kafka消息生产者
spring.kafka.four.enabled=true
spring.kafka.four.producer.name=fourKafkaSender
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

国籍惯例,先上源码: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生产者

二、开发环境

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

三、原项目

1、接前文,我们基本完成了kafka consumer常用的特性开发,有小伙伴问,我们该如何配置多个数据源生产者,想consumer一样简单,发送kafka消息呢?


## 1.配置
spring.kafka.four.enabled=true
spring.kafka.four.producer.name=fourKafkaSender
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## 2.引用
@Resource(name = "fourKafkaSender")
private MmcKafkaMultiSender mmcKafkaMultiSender;## 3.使用
mmcKafkaMultiSender.sendStringMessage(topicOne, "aaa", json);

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

四、修改项目

1、修改内部类MmcKafkaProperties类,增加生产者相关的配置。

    @EqualsAndHashCode(callSuper = true)@Datapublic static class Producer extends KafkaProperties.Producer {/*** 是否启用.*/private boolean enabled = true;/*** 生产者名称,如果有设置则会覆盖默认的xxxKakfkaSender名称.*/private String name;}/*** 生产者.*/private final Producer producer = new Producer();/*** Create an initial map of producer properties from the state of this instance.* <p>* This allows you to add additional properties, if necessary, and override the* default kafkaProducerFactory bean.** @return the producer properties initialized with the customizations defined on this*         instance*/Map<String, Object> buildProducerProperties() {return new HashMap<>(this.producer.buildProperties());}

2、新增MmcKafkaSender接口,作为发送Kafka消息的唯一约束。

public interface MmcKafkaSender {/*** 发送kafka消息.** @param topic        topic名称* @param partitionKey 消息分区键* @param message      具体消息*/void sendStringMessage(String topic, String partitionKey, String message);/*** 发送kafka消息.** @param topic        topic名称* @param partitionKey 消息分区键* @param message      具体消息*/void sendProtobufMessage(String topic, String partitionKey, byte[] message);
}

3、新增MmcKafkaOutputContainer容器类,用于存储所有生产者,方便统一管理;

@Getter
@Slf4j
public class MmcKafkaOutputContainer {/*** 存放所有生产者.*/private final Map<String, MmcKafkaSender> outputs;/*** 构造函数.*/public MmcKafkaOutputContainer(Map<String, MmcKafkaSender> outputs) {this.outputs = outputs;}}

4、新增MmcKafkaSingleSender实现类,用于真实发送Kafka消息;

public class MmcKafkaSingleSender implements MmcKafkaSender {private final KafkaTemplate<String, Object> template;public MmcKafkaSingleSender(KafkaTemplate<String, Object> template) {this.template = template;}@Overridepublic void sendStringMessage(String topic, String partitionKey, String message) {template.send(topic, partitionKey, message);}@Overridepublic void sendProtobufMessage(String topic, String partitionKey, byte[] message) {template.send(topic, partitionKey, message);}}

5、修改MmcMultiProducerAutoConfiguration配置类,遍历所有配置,组装并生成MmcKafkaSingleSender,并注册到IOC容器;


@Slf4j
@Configuration
@EnableConfigurationProperties(MmcMultiKafkaProperties.class)
@ConditionalOnProperty(prefix = "spring.kafka", value = "enabled", matchIfMissing = true)
public class MmcMultiProducerAutoConfiguration implements BeanFactoryAware {private DefaultListableBeanFactory beanDefinitionRegistry;@Resourceprivate MmcMultiKafkaProperties mmcMultiKafkaProperties;@Beanpublic MmcKafkaOutputContainer mmcKafkaOutputContainer() {// 初始化一个存储所有生产者的哈希映射Map<String, MmcKafkaSender> outputs = new HashMap<>();// 获取所有的Kafka配置信息Map<String, MmcMultiKafkaProperties.MmcKafkaProperties> kafkas = mmcMultiKafkaProperties.getKafka();// 逐个遍历,并生成producerfor (Map.Entry<String, MmcMultiKafkaProperties.MmcKafkaProperties> entry : kafkas.entrySet()) {// 唯一生产者名称String name = entry.getKey();// 生产者配置MmcMultiKafkaProperties.MmcKafkaProperties properties = entry.getValue();// 是否开启if (properties.isEnabled()&& properties.getProducer().isEnabled()&& CommonUtil.isNotEmpty(properties.getProducer().getBootstrapServers())) {// bean名称String beanName = Optional.ofNullable(properties.getProducer().getName()).orElse(name + "KafkaSender");KafkaTemplate<String, Object> template = mmcdKafkaTemplate(properties);// 创建实例MmcKafkaSender sender = new MmcKafkaSingleSender(template);outputs.put(beanName, sender);// 注册到IOCbeanDefinitionRegistry.registerSingleton(beanName, sender);}}return new MmcKafkaOutputContainer(outputs);}private KafkaTemplate<String, Object> mmcdKafkaTemplate(MmcMultiKafkaProperties.MmcKafkaProperties producer) {return new KafkaTemplate<>(baseKafkaProducerFactory(producer));}private ProducerFactory<String, Object> baseKafkaProducerFactory(MmcMultiKafkaProperties.MmcKafkaProperties producer) {return new DefaultKafkaProducerFactory<>(producer.buildProducerProperties());}@Overridepublic void setBeanFactory(BeanFactory beanFactory) throws BeansException {this.beanDefinitionRegistry = (DefaultListableBeanFactory) beanFactory;}
}

五、测试一下

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、消费者配置保持不变,增加生产者配置。

## json消息消费者
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=oneProcessor
spring.kafka.one.duplicate=false
spring.kafka.one.snakeCase=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
spring.kafka.one.container.threshold=2
spring.kafka.one.container.rate=1000
spring.kafka.one.container.parallelism=8## json消息生产者
spring.kafka.four.enabled=true
spring.kafka.four.producer.name=fourKafkaSender
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

3、编写测试类。

@Slf4j
@ActiveProfiles("dev")
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = {MmcMultiProducerAutoConfiguration.class, MmcMultiConsumerAutoConfiguration.class,DemoService.class, OneProcessor.class})
@TestPropertySource(value = "classpath:application-string.properties")
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"},topics = {"${spring.kafka.one.topic}"})
class KafkaStringMessageTest {@Value("${spring.kafka.one.topic}")private String topicOne;@Value("${spring.kafka.two.topic}")private String topicTwo;@Resource(name = "fourKafkaSender")private MmcKafkaSingleSender mmcKafkaSingleSender;@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);mmcKafkaSingleSender.sendStringMessage(topicOne, "aaa", json);}}
}

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

五、小结

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

《搭建大型分布式服务(三十六)SpringBoot 零代码方式整合多个kafka数据源》
《搭建大型分布式服务(三十七)SpringBoot 整合多个kafka数据源-取消限定符》
《搭建大型分布式服务(三十八)SpringBoot 整合多个kafka数据源-支持protobuf》
《搭建大型分布式服务(三十九)SpringBoot 整合多个kafka数据源-支持Aware模式》
搭建大型分布式服务(四十)SpringBoot 整合多个kafka数据源-支持生产者
搭建大型分布式服务(四十一)SpringBoot 整合多个kafka数据源-支持亿级消息生产者

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

这篇关于搭建大型分布式服务(四十)SpringBoot 整合多个kafka数据源-支持生产者的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1088294

相关文章

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时,有没有考虑过 JVM 是如何把静态的字节码(byte code)转化为运行时对象的呢,这个问题看似简单,但清楚的同学相信也不会太多,这篇文章首先介绍 JVM 类初始化的机制,然后给出几个易出错的实例来分析,帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段,分别是:loading 、Linking、initialization

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权,允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true (Spring security默认会带有ROLE_前缀),去

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程,特意画了如下的类图,包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口,接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类,它们之间的依赖,构建起了整个框架。想要理解整个架构,最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文(security context)的信息。当前操作的用户是谁,该用户是否已经被认证,他拥有哪些角色权限…这些都被保

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明(认真看哦) AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证,则会调用 #attemptAuthentica

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门(一):登录与退出》 《Spring-Security-入门(二):基于数据库验证》 《Spring-Security-入门(三):密码加密》 《Spring-Security-入门(四):自定义-Filter》 《Spring-Security-入门(五):在 Sprin

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物

Java进阶13讲__第12讲_1/2

多线程、线程池 1.  线程概念 1.1  什么是线程 1.2  线程的好处 2.   创建线程的三种方式 注意事项 2.1  继承Thread类 2.1.1 认识  2.1.2  编码实现  package cn.hdc.oop10.Thread;import org.slf4j.Logger;import org.slf4j.LoggerFactory

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来,一站式有声阅读平台听书系统 🌟&nbsp;开篇:遇见未来,从“智听”开始 在这个快节奏的时代,你是否渴望在忙碌的间隙,找到一片属于自己的宁静角落?是否梦想着能随时随地,沉浸在知识的海洋,或是故事的奇幻世界里?今天,就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统,它正悄悄改变着我们的阅读方式,让未来触手可及! 📚&nbsp;第一站:海量资源,应有尽有 走进“智听

搭建Kafka+zookeeper集群调度

前言 硬件环境 172.18.0.5        kafkazk1        Kafka+zookeeper                Kafka Broker集群 172.18.0.6        kafkazk2        Kafka+zookeeper                Kafka Broker集群 172.18.0.7        kafkazk3