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

系列文章目录

前言

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

<dependency>
    <groupId>io.github.vipjoey</groupId>
    <artifactId>multi-kafka-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

## 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、接前文，我们整合生产者到这个组件，那么假如我们生产的消息过亿级别，一个生产者不足以支持这么大的buffer发送量级，我们该如何操作？是否支持配置多个数量的生产者？

## 1.配置
spring.kafka.four.enabled=true
spring.kafka.four.producer.count=1 ## 生产者数量，默认为1个
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 MmcKafkaSender mmcKafkaMultiSender;

## 3.使用
mmcKafkaMultiSender.sendStringMessage(topicOne, "aaa", json);

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

四、修改项目

1、新增MmcKafkaMultiSender类，用于支持任意数量的发送者，利用负载均衡发送消息。

public class MmcKafkaMultiSender implements MmcKafkaSender {

    private final AtomicLong atomicLong = new AtomicLong(1);

    private final List<KafkaTemplate<String, Object>> templates;

    public MmcKafkaMultiSender(List<KafkaTemplate<String, Object>> templates) {
        this.templates = templates;
    }

    @Override
    public void sendStringMessage(String topic, String partitionKey, String message) {

        KafkaTemplate<String, Object> template = templates.get((int) (atomicLong.getAndIncrement() % templates.size()));
        template.send(topic, partitionKey, message);
    }


    @Override
    public void sendProtobufMessage(String topic, String partitionKey, byte[] message) {

        KafkaTemplate<String, Object> template = templates.get((int) (atomicLong.getAndIncrement() % templates.size()));
        template.send(topic, partitionKey, message);

    }
}

2、修改MmcMultiProducerAutoConfiguration配置类，配置多个数量的Sender；

 @Bean
    public MmcKafkaOutputContainer mmcKafkaOutputContainer() {

        // 初始化一个存储所有生产者的哈希映射
        Map<String, MmcKafkaSender> outputs = new HashMap<>();

        // 获取所有的Kafka配置信息
        Map<String, MmcMultiKafkaProperties.MmcKafkaProperties> kafkas = mmcMultiKafkaProperties.getKafka();

        // 逐个遍历，并生成producer
        for (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");


                // 数量
                List<KafkaTemplate<String, Object>> templates = new ArrayList<>(properties.getProducer().getCount());
                for (int i = 0; i < properties.getProducer().getCount(); i++) {

                    log.info("[pando] init producer {} - {} ", name, i);
                    KafkaTemplate<String, Object> template = mmcKafkaTemplate(properties);
                    templates.add(template);
                }

                // 创建实例
                MmcKafkaSender sender = new MmcKafkaMultiSender(templates);
                outputs.put(beanName, sender);

                // 注册到IOC
                beanDefinitionRegistry.registerSingleton(beanName, sender);
            }

        }

        return new MmcKafkaOutputContainer(outputs);
    }

五、测试一下

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 MmcKafkaSender mmcKafkaSender;


    @Test
    void 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);

            mmcKafkaSender.sendStringMessage(topicOne, "aaa", json);


        }
    }
}

5、运行一下，测试通过，可以看到能正常发送消息和消费。

五、小结

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

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

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