使用 spring kafka 2.5.8 版本实现每个 kafka 主题分区一个消费者线程

Question

我一直在使用 apache kafka-clients（准确地说是 2.3.1 版本）库来创建 kafka 消费者，其中一个分区 - 一个消费者线程是通过以下计算实现的：

计算上的消费者线程数 * 计算数 = 主题的分区数

过去是手动缩放，因此当需要增加计算数量时，一个计算上运行的消费者线程的数量会相应减少。

我们如何使用org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory 来实现这一点。

我正在尝试使用 spring kafka 2.5.8 版本。该应用程序在具有自动扩展能力的 k8s 集群上运行。假设我将最大和最小 pod 设置为 4，那么理想情况下

4 X 消费者线程数 = 主题的分区数

这个消费者线程数是如何配置的。是不是通过这个：

org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory.setConcurrency() 。有人可以指导一下吗？

Answer 1

是的；或 @KafkaListener 上的 concurrency 属性覆盖工厂的并发性。

如果你在运行时更改它，它不会生效，除非你stop()和start()容器。