Spring Kafka中多个ConcurrentMessageListenerContainer能否共享公共线程池

该需求在Spring Kafka中可以直接实现，不需要自行实例化KafkaConsumers做自定义开发。

• 首先初始化公共线程池，你可以根据业务量级自定义线程池参数，推荐使用Spring封装的ThreadPoolTaskExecutor实现：

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;

// 注入为公共Bean
@Bean
public Executor kafkaCommonConsumeExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    // 核心线程数建议匹配日常消费需要的最低并发数
    executor.setCorePoolSize(30);
    // 最大线程数建议大于所有消费者容器concurrency参数的总和
    executor.setMaxPoolSize(80);
    executor.setQueueCapacity(200);
    executor.setThreadNamePrefix("kafka-consume-");
    executor.initialize();
    return executor;
}

• 给每个ConcurrentMessageListenerContainer实例绑定同一个公共线程池，只需要修改容器配置的taskExecutor属性即可：

@Bean
public ConcurrentMessageListenerContainer<String, Object> topic1Container(
        ConcurrentKafkaListenerContainerFactory<String, Object> factory,
        Executor kafkaCommonConsumeExecutor) {
    ConcurrentMessageListenerContainer<String, Object> container = 
            factory.createContainer("topic1");
    container.getContainerProperties().setGroupId("consumer-group1");
    // 绑定公共线程池
    container.getContainerProperties().setTaskExecutor(kafkaCommonConsumeExecutor);
    // 其他自定义配置...
    return container;
}
// 其余topic对应的消费者容器都复用同一个kafkaCommonConsumeExecutor即可

• 如果需要统一管控kafkaConsumer.poll的调用逻辑和消息处理流程，可以额外配置全局的RecordInterceptor或者ConsumerInterceptor，在拦截器中实现统一的逻辑埋点，不需要修改每个监听器的业务代码。

• 公共线程池的最大线程数不要设置过小，要预留足够的冗余应对并发消费峰值，避免线程资源争抢导致消费延迟。
• 如果需要对不同消费者组做资源用量的监控，可以给线程池加上自定义的线程标识，统计各消费者组的线程占用情况。

内容的提问来源于stack exchange，提问作者loogies