系列文章目录

---

---

前言

在过去的一段时间里，我们利用了AI大模型写了一个多线程并发框架，那么，我们怎样集成到Kafka组件里，让消费速度提升N倍呢？

• 《AI大模型编写多线程并发框架（六十一）：从零开始搭建框架》
• 《AI大模型编写多线程并发框架（六十二）：限流和并发度优化》
• 《AI大模型编写多线程并发框架（六十三）：监听器优化·上》
• 《AI大模型编写多线程并发框架（六十四）：监听器优化·下》
• 《AI大模型编写多线程并发框架（六十五）：发布和应用》

国籍惯例，先上源码：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生产者
• SpringBoot Kafka单次batch消息内并发处理

二、开发环境

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

三、原项目

1、接前文，我们已经发布了Kafka组件到中央仓库，所有开发者都可以下载使用本组件。虽然本组件支持批量消费Kafka消息，但是毕竟它是串行顺序处理的，尤其涉及高IO耗时调用时，比如消费Kafka，然后读写DB多表操作这种场景，会使消费速度下降。能否并发处理这些Kafka消息呢？

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

四、修改项目

1、新增ContainerConfig接口类，用于获取多线程任务容器配置，便于后续使用Apollo、Disconf、Consul等配置中心。

public interface ContainerConfig {

    /**
     * Get the execute rate.
     *
     * @return rate
     */
    int getRate();

    /**
     * Get the max task count for per thread.
     *
     * @return max count
     */
    int getThreshold();


    /**
     * The max thread count, default is numbers of processor.
     * @return count
     */
    default int getParallelism() {
        return Runtime.getRuntime().availableProcessors();
    }
}

2、修改MmcMultiKafkaProperties配置类，增加容器配置；

@ToString
@Data
@ConfigurationProperties(prefix = "spring")
public class MmcMultiKafkaProperties {

	// 省略其他代码
	
    /**
     * MmcKafkaProperties.
     */
    @Data
    static class MmcKafkaProperties {

  		// 省略其他代码
  		
        /**
         * 并发设置.
         */
        private Container container = new Container();

    }

    @Data
    public static class Container implements ContainerConfig {

        /**
         * 是否启用多线程消费.
         */
        private boolean enabled = true;
        /*
         * 消费消息的速率（每秒接收的记录数），默认值为1000.
         */
        private int rate = 1000;
        /*
         * 最小批次数量，默认为2.
         */
        private int threshold = 2;
        /*
         * 设置并行度，默认值为可用处理器数量.
         */
        private int parallelism = Runtime.getRuntime().availableProcessors();
    }
}

3、修改MmcMultiConsumerAutoConfiguration配置类，主要是增加inputer的初始化方法，用于后续构建多线程任务容器实例。

public interface MmcInputer {

	// 省略其他代码
	
    /**
     * 初始化kafka容器.
     */
    void init();
}


@Slf4j
@Configuration
@EnableConfigurationProperties(MmcMultiKafkaProperties.class)
@ConditionalOnProperty(prefix = "spring.kafka", value = "enabled", matchIfMissing = true)
public class MmcMultiConsumerAutoConfiguration extends BaseConsumerConfiguration {

   // 省略其他代码

    @Bean
    public MmcKafkaInputerContainer mmcKafkaInputerContainer(MmcKafkaProcessorFactory factory,
                                                             MmcKafkaBeanPostProcessor beanPostProcessor) throws Exception {

        // 省略其他代码

        // 逐个遍历，并生成consumer
        for (Map.Entry<String, MmcMultiKafkaProperties.MmcKafkaProperties> entry : kafkas.entrySet()) {

			// 省略其他代码

            // 是否开启
            if (properties.isEnabled() && CommonUtil.isNotBlank(properties.getGroupId())) {

                // 省略其他代码
                
                // 设置容器
                inputer.setContainer(container);
                inputer.setName(name);
                inputer.setProperties(properties);
                inputer.init(); // 增加初始化

				// 省略其他代码
               
            }

        }

        return new MmcKafkaInputerContainer(inputers);
    }
}

4、由于增加了inputer增加了init方法，所以超级父类KafkaAbstractProcessor也增加一个默认实现。

@Slf4j
@Setter
public abstract class KafkaAbstractProcessor<T> implements MmcInputer {

	// 省略其他代码
	
    @Override
    public void init() {

    }
}

5、新增MmcKafkaParallelAbstractProcessor并发处理类，根据多线程并发框架mmc-juc的特性，配置初始化多线程任务容器，并保留很多回调函数，方便子类覆盖重写。

@Slf4j
@Setter
public abstract class MmcKafkaParallelAbstractProcessor<T, R> extends MmcKafkaAbstractProcessor<T> {

    /**
     * taskExecutor.
     */
    protected MmcTaskExecutor<T, R> taskExecutor;

    /**
     * init.
     */
    public void init() {

        ContainerConfig config = properties.getContainer();

        this.taskExecutor = MmcTaskExecutor.<T, R>builder()
                .taskProcessor(this::handelBatchDatas)
                .threshold(config.getThreshold())
                .rateLimiter(buildRateLimiter(config.getRate()))
                .taskMerger(this::mergeResult)
                .forkJoinPoolConcurrency(config.getParallelism())
                .build();
    }


    @Override
    protected void dealMessage(List<T> datas) throws ExecutionException, InterruptedException {

        if (properties.getContainer().isEnabled()) {

            // 开启并发处理
            R result = taskExecutor.execute(MmcTask.<T, R>builder()
                    .taskSource(datas)
                    .taskName(getTaskName(datas))
                    .build()
            );

            dealMessageCallBack(result);

        } else {

            // 同步处理
            R result = handelBatchDatas(datas);
            dealMessageCallBack(result);

        }
    }

    /**
     * 合并小任务结果（默认不合并）.
     *
     * @param left 左边处理结果
     * @param right 右边处理结果
     * @return 合并后的结果
     */
    protected R mergeResult(R left, R right) {
        return null;
    }

    /**
     * 构建速率限制器.
     *
     * @param rate qps
     * @return 速率限制器
     */
    protected RateLimiter buildRateLimiter(int rate) {
        return new TokenBucket(rate, rate);
    }

    /**
     * 当所有消息处理完后，会调用该方法.
     *
     * @param result 处理结果
     */
    protected void dealMessageCallBack(R result) {
        // default null
    }

    /**
     * 获取任务名称.
     */
    protected String getTaskName(List<T> datas) {
        return name;
    }


    /**
     * 真正处理消息的方法.
     *
     * @param datas 待处理消息
     * @return 小任务处理完的结果
     */
    protected abstract R handelBatchDatas(List<T> datas);

}

五、测试一下

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

       <dependency>
	    <groupId>io.github.vipjoey</groupId>
	    <artifactId>mmc-juc</artifactId>
	    <version>1.1</version>
	</dependency>

2、增加并发消费者配置，生产者配置不变。

## json消息消费者
spring.kafka.five.enabled=true
spring.kafka.five.consumer.bootstrapServers=${spring.embedded.kafka.brokers}
spring.kafka.five.topic=mmc-topic-five
spring.kafka.five.group-id=group-consumer-five
spring.kafka.five.processor=fiveProcessor
spring.kafka.five.duplicate=true
spring.kafka.five.snakeCase=false
spring.kafka.five.consumer.auto-offset-reset=latest
spring.kafka.five.consumer.max-poll-records=10
spring.kafka.five.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.five.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
## 并发配置
spring.kafka.five.container.threshold=2
spring.kafka.five.container.rate=1000
spring.kafka.five.container.parallelism=8

## json消息生产者
spring.kafka.five.enabled=true
spring.kafka.five.producer.name=fiveKafkaSender
spring.kafka.five.producer.bootstrap-servers=${spring.embedded.kafka.brokers}
spring.kafka.five.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.five.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

3、编写测试类。

@Slf4j
@Service("fiveProcessor")
public class FiveProcessor extends MmcKafkaParallelAbstractProcessor<ParalleMsg, Void> {


    @Override
    protected Void handelBatchDatas(List<ParalleMsg> datas) {
        datas.forEach(x -> {
            log.info("handelBatchDatas one: {}", x);
        });

        return null;
    }
}

@Slf4j
@ActiveProfiles("dev")
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = {MmcMultiProducerAutoConfiguration.class, MmcMultiConsumerAutoConfiguration.class,
        FiveProcessor.class})
@TestPropertySource(value = "classpath:application-paralle.properties")
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"},
        topics = {"${spring.kafka.five.topic}"})
public class KafkaParalleMessageTest {


    @Value("${spring.kafka.five.topic}")
    private String fiveTopic;

    @Resource(name = "fiveKafkaSender")
    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(fiveTopic, "aaa", json);


        }
    }
}

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

五、小结

将本项目代码构建成starter，就可以大大提升我们开发效率，我们只需要关心业务代码的开发，github项目源码：轻触这里。如果对你有用可以打个星星哦。

• 《搭建大型分布式服务（三十六）SpringBoot 零代码方式整合多个kafka数据源》
• 《搭建大型分布式服务（三十七）SpringBoot 整合多个kafka数据源-取消限定符》
• 《搭建大型分布式服务（三十八）SpringBoot 整合多个kafka数据源-支持protobuf》
• 《搭建大型分布式服务（三十九）SpringBoot 整合多个kafka数据源-支持Aware模式》
• 《搭建大型分布式服务（四十）SpringBoot 整合多个kafka数据源-支持生产者》
• 《搭建大型分布式服务（四十一）SpringBoot 整合多个kafka数据源-支持亿级消息生产者》
• 《搭建大型分布式服务（四十二）SpringBoot 无代码侵入实现多Kafka数据源整合插件发布》
• 《搭建大型分布式服务（四十三）SpringBoot 多Kafka数据源发布到Maven中央仓库：让世界看到你的作品！》
• 《搭建大型分布式服务（四十四）SpringBoot 无代码侵入实现多Kafka数据源：单分区提升至十万级消费速度！》

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