0
0

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

系列文章目录

文章目录

前言

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

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

例如下面这样简单的配置就完成SpringBoot和kafka的整合，我们只需要关心

  1. com.mmc.multi.kafka.starter.OneProcessor

和

  1. com.mmc.multi.kafka.starter.TwoProcessor

这两个Service的代码开发。

  1. ## topic1的kafka配置
  2. spring.kafka.one.enabled=true
  3. spring.kafka.one.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
  4. spring.kafka.one.topic=mmc-topic-one
  5. spring.kafka.one.group-id=group-consumer-one
  6. spring.kafka.one.processor=com.mmc.multi.kafka.starter.OneProcessor // 业务处理类名称
  7. spring.kafka.one.consumer.auto-offset-reset=latest
  8. spring.kafka.one.consumer.max-poll-records=10
  9. spring.kafka.one.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  10. spring.kafka.one.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  12. ## topic2的kafka配置
  13. spring.kafka.two.enabled=true
  14. spring.kafka.two.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
  15. spring.kafka.two.topic=mmc-topic-two
  16. spring.kafka.two.group-id=group-consumer-two
  17. spring.kafka.two.processor=com.mmc.multi.kafka.starter.TwoProcessor // 业务处理类名称
  18. spring.kafka.two.consumer.auto-offset-reset=latest
  19. spring.kafka.two.consumer.max-poll-records=10
  20. spring.kafka.two.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  21. spring.kafka.two.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  23. ## pb 消息消费者
  24. spring.kafka.pb.enabled=true
  25. spring.kafka.pb.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
  26. spring.kafka.pb.topic=mmc-topic-pb
  27. spring.kafka.pb.group-id=group-consumer-pb
  28. spring.kafka.pb.processor=pbProcessor
  29. spring.kafka.pb.consumer.auto-offset-reset=latest
  30. spring.kafka.pb.consumer.max-poll-records=10
  31. spring.kafka.pb.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  32. spring.kafka.pb.consumer.value-deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
  34. ## kafka消息生产者
  35. spring.kafka.four.enabled=true
  36. spring.kafka.four.producer.name=fourKafkaSender
  37. spring.kafka.four.producer.bootstrap-servers=${spring.embedded.kafka.brokers}
  38. spring.kafka.four.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  39. 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消息呢？

  2. ## 1.配置
  3. spring.kafka.four.enabled=true
  4. spring.kafka.four.producer.name=fourKafkaSender
  5. spring.kafka.four.producer.bootstrap-servers=${spring.embedded.kafka.brokers}
  6. spring.kafka.four.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  7. spring.kafka.four.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
  9. ## 2.引用
  10. @Resource(name ="fourKafkaSender")privateMmcKafkaMultiSender mmcKafkaMultiSender;
  12. ## 3.使用
  13. mmcKafkaMultiSender.sendStringMessage(topicOne,"aaa", json);

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

四、修改项目

1、修改内部类

  1. MmcKafkaProperties

类，增加生产者相关的配置。

  1. @EqualsAndHashCode(callSuper =true)@DatapublicstaticclassProducerextendsKafkaProperties.Producer{/**
  2.          * 是否启用.
  3.          */privateboolean enabled =true;/**
  4.          * 生产者名称，如果有设置则会覆盖默认的xxxKakfkaSender名称.
  5.          */privateString name;}/**
  6.          * 生产者.
  7.          */privatefinalProducer producer =newProducer();/**
  8.          * Create an initial map of producer properties from the state of this instance.
  9.          * <p>
  10.          * This allows you to add additional properties, if necessary, and override the
  11.          * default kafkaProducerFactory bean.
  12.          *
  13.          * @return the producer properties initialized with the customizations defined on this
  14.          *         instance
  15.          */Map<String,Object>buildProducerProperties(){returnnewHashMap<>(this.producer.buildProperties());}

2、新增

  1. MmcKafkaSender

接口，作为发送Kafka消息的唯一约束。

  1. publicinterfaceMmcKafkaSender{/**
  2.      * 发送kafka消息.
  3.      *
  4.      * @param topic        topic名称
  5.      * @param partitionKey 消息分区键
  6.      * @param message      具体消息
  7.      */voidsendStringMessage(String topic,String partitionKey,String message);/**
  8.      * 发送kafka消息.
  9.      *
  10.      * @param topic        topic名称
  11.      * @param partitionKey 消息分区键
  12.      * @param message      具体消息
  13.      */voidsendProtobufMessage(String topic,String partitionKey,byte[] message);}

3、新增

  1. MmcKafkaOutputContainer

容器类，用于存储所有生产者，方便统一管理；

  1. @Getter@Slf4jpublicclassMmcKafkaOutputContainer{/**
  2.      * 存放所有生产者.
  3.      */privatefinalMap<String,MmcKafkaSender> outputs;/**
  4.      * 构造函数.
  5.      */publicMmcKafkaOutputContainer(Map<String,MmcKafkaSender> outputs){this.outputs = outputs;}}

4、新增

  1. MmcKafkaSingleSender

实现类，用于真实发送Kafka消息；

  1. publicclassMmcKafkaSingleSenderimplementsMmcKafkaSender{privatefinalKafkaTemplate<String,Object> template;publicMmcKafkaSingleSender(KafkaTemplate<String,Object> template){this.template = template;}@OverridepublicvoidsendStringMessage(String topic,String partitionKey,String message){
  3.         template.send(topic, partitionKey, message);}@OverridepublicvoidsendProtobufMessage(String topic,String partitionKey,byte[] message){
  5.         template.send(topic, partitionKey, message);}}

5、修改

  1. MmcMultiProducerAutoConfiguration

配置类，遍历所有配置，组装并生成

  1. MmcKafkaSingleSender

，并注册到IOC容器；

  1. @Slf4j@Configuration@EnableConfigurationProperties(MmcMultiKafkaProperties.class)@ConditionalOnProperty(prefix ="spring.kafka", value ="enabled", matchIfMissing =true)publicclassMmcMultiProducerAutoConfigurationimplementsBeanFactoryAware{privateDefaultListableBeanFactory beanDefinitionRegistry;@ResourceprivateMmcMultiKafkaProperties mmcMultiKafkaProperties;@BeanpublicMmcKafkaOutputContainermmcKafkaOutputContainer(){// 初始化一个存储所有生产者的哈希映射Map<String,MmcKafkaSender> outputs =newHashMap<>();// 获取所有的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 =newMmcKafkaSingleSender(template);
  2.                 outputs.put(beanName, sender);// 注册到IOC
  3.                 beanDefinitionRegistry.registerSingleton(beanName, sender);}}returnnewMmcKafkaOutputContainer(outputs);}privateKafkaTemplate<String,Object>mmcdKafkaTemplate(MmcMultiKafkaProperties.MmcKafkaProperties producer){returnnewKafkaTemplate<>(baseKafkaProducerFactory(producer));}privateProducerFactory<String,Object>baseKafkaProducerFactory(MmcMultiKafkaProperties.MmcKafkaProperties producer){returnnewDefaultKafkaProducerFactory<>(producer.buildProducerProperties());}@OverridepublicvoidsetBeanFactory(BeanFactory beanFactory)throwsBeansException{this.beanDefinitionRegistry =(DefaultListableBeanFactory) beanFactory;}}

五、测试一下

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

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

  1. ## json消息消费者
  2. spring.kafka.one.enabled=true
  3. spring.kafka.one.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
  4. spring.kafka.one.topic=mmc-topic-one
  5. spring.kafka.one.group-id=group-consumer-one
  6. spring.kafka.one.processor=oneProcessor
  7. spring.kafka.one.duplicate=false
  8. spring.kafka.one.snakeCase=false
  9. spring.kafka.one.consumer.auto-offset-reset=latest
  10. spring.kafka.one.consumer.max-poll-records=10
  11. spring.kafka.one.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  12. spring.kafka.one.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  13. spring.kafka.one.container.threshold=2
  14. spring.kafka.one.container.rate=1000
  15. spring.kafka.one.container.parallelism=8
  17. ## json消息生产者
  18. spring.kafka.four.enabled=true
  19. spring.kafka.four.producer.name=fourKafkaSender
  20. spring.kafka.four.producer.bootstrap-servers=${spring.embedded.kafka.brokers}
  21. spring.kafka.four.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  22. spring.kafka.four.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3、编写测试类。

  1. @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"},
  2.         topics ={"${spring.kafka.one.topic}"})classKafkaStringMessageTest{@Value("${spring.kafka.one.topic}")privateString topicOne;@Value("${spring.kafka.two.topic}")privateString topicTwo;@Resource(name ="fourKafkaSender")privateMmcKafkaSingleSender mmcKafkaSingleSender;@TestvoidtestDealMessage()throwsException{Thread.sleep(2*1000);// 模拟生产数据produceMessage();Thread.sleep(10*1000);}voidproduceMessage(){for(int i =0; i <10; i++){DemoMsg msg =newDemoMsg();
  3.             msg.setRoutekey("routekey"+ i);
  4.             msg.setName("name"+ i);
  5.             msg.setTimestamp(System.currentTimeMillis());String json =JsonUtil.toJsonStr(msg);
  7.             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数据源整合插件发布》
  • 《搭建大型分布式服务（四十三）SpringBoot 多Kafka数据源发布到Maven中央仓库：让世界看到你的作品！》

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

标签： spring boot kafka linq
本文转载自: https://blog.csdn.net/hanyi_/article/details/139897315
版权归原作者 hanyi_ 所有， 如有侵权，请联系我们删除。
登录后发布评论

“搭建大型分布式服务（四十）SpringBoot 整合多个kafka数据源-支持生产者”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？