Kafka 并发配置 Spring setConsumerTaskExecutor 和 setConcurrency

Question

spring kafka 中的 setConsumerTaskExecutor() 和 setConcurrency() 有什么区别？ 如果 ConsumerTaskExecutor 的 maxpool 大小与 setConcurrency 值不同会发生什么？

ThreadPoolTaskExecutor customExecutor= new ThreadPoolTaskExecutor();
exec.setCorePoolSize(3);
exec.setMaxPoolSize(6);

ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<String, String>();

factory.setConcurrency(10);
factory.getContainerProperties().setConsumerTaskExecutor(customExecutor);

谢谢

Answer 1

concurrency:

同时运行的 KafkaMessageListenerContainer 的最大数量。来自同一分区内的消息将按顺序处理。

ConsumerTaskExecutor:

为轮询消费者的线程设置执行器。

https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.html#getConcurrency()

https://docs.spring.io/spring-kafka/docs/current/api/org/springframework/kafka/listener/ContainerProperties.html#setConsumerTaskExecutor(org.springframework.core.task.AsyncListenableTaskExecutor)

---

换句话说，ConcurrentKafkaListenerContainer 基于concurrency 创建一个或多个KafkaMessageListenerContainers。每个KafkaMessageListenerContainer 共享ContainerProperties，使用ConsumerTaskExecutor 运行ListenerConsumer，它委托给Kafka consumer 进行投票。