Spring Kafka MessageListenerContainer答案

【问题标题】：Spring Kafka MessageListenerContainerSpring Kafka MessageListenerContainer
【发布时间】：2019-07-10 16:17:54
【问题描述】：

我正在看spring Kafka代码，我有一些疑问：

如果我们使用 1 个 @kafkaListener 和 2 个主题，那么 spring Kafka 会创建一个 MessageListenerContainer。如果我为每个主题使用单独的@kafkaListener，则会创建 2 个 MessageListenerContainer。
MessageListenerContainer 是指消费者吗？
如果我在 ConcurrentKafkaListenerContainerFactory 中将并发设置为 4，那么这意味着对于每个 kafkaListener，我使用代理打开 4 个线程？这意味着协调员将他们视为 4 个不同的消费者。
轮询如何与 kafkaListener 一起工作？每次只能从 broker 获得 1 个 ConsumerRecord 吗？

请帮忙。

【问题讨论】：

标签： apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

MessageListenerContainer 有两种实现方式 - KafkaMessageListenerContainer (KMLC) 和 ConcurrentMessageListenerContainer (CMLC)。

CMLC 只是一个或多个 KMLC 的包装器，KMLC 的数量由 concurrency 指定。

@KafkaListener 始终使用 CMLC。

每个 KMLC 都有一个 Consumer（和一个线程）。线程不断poll()s 消费者，指定pollTimeout。

主题/分区在 KMLC 中的分布方式取决于

主题有多少个分区
消费者的partition.assignment.strategy属性

如果您有多个分区少于并发的主题，您可能需要一个备用分区分配器，例如循环分配器，否则您将有没有分配的空闲容器。

这是正确的；如果您明确希望每个主题使用不同的容器，则可以在同一方法上提供多个 @KafkaListener 注释。
见我上面的解释。
这是正确的 - 这是与 Kafka 实现并发的唯一方法（无需添加非常复杂的逻辑来管理偏移量）。
每次轮询返回的记录数取决于消费者属性的数量，max.poll.records、fetch.min.bytes、fetch.max.wait.ms。

【讨论】：

谢谢@Gary。现在在 Kafka 客户端 API 中，消费者是使用来自 Kafka 集群的记录的客户端。如果我使用的是 Spring Kafka，并且我有 2 个 KMLC，这意味着有 2 个消费者连接到集群，每个 KMLC 连接到不同的客户端。
另外，假设我有一个有 10 个分区的主题 A，我的并发是 5。所以这将创建 5 个消费者进程，每个进程将从 2 个不同的分区消费。现在，如果我使用相同的 conf 启动相同的应用程序。在另一台机器上，所以对于 Kafka 协调器来说，它看起来像 10 个不同的消费者，所以现在所有消费者将分别获得 1 个分区，在这种情况下，一台消费者机器将获得 5 个分区。现在，如果我将此并发增加到 10，是否有可能其中一台消费者机器没有分配任何分区并处于空闲状态。
没错。如果消费者具有相同的group.id 属性，则分区将在它们之间分配。如果他们有不同的group.ids，他们都会收到所有记录。 CLMC 的子 KMLC 都获得相同的group.id。是的;您的分区数必须 >= number of instances * concurrency 以避免空闲消费者。
那么你如何处理并发，如果我订阅的主题的分区数量少于并发，因为在创建消费者时，如果我知道你不知道分区订阅主题。我已经看到，在 TopicPartition 的情况下，并发性会减少到 topicPartition 的长度。因此，如果我有一个有 10 个分区的主题 A 并且我的并发性是 11，那么第 11 个消费者线程会发生什么。这会向代理发送任何心跳，还是会在 session.timeout.ms 之后导致重新平衡？
空闲消费者继续在其循环中轮询消费者。它永远不会得到任何记录。这样，如果发生再平衡，它可以参与（分区分配发生在poll() 序列内）。假设您有 10 个分区、2 个实例和 6 个并发。当两个实例都处于活动状态时，您将有 2 个空闲消费者。如果一个实例停止，则 10 个分区将分布在其余 6 个消费者（包括空闲的消费者）中。此外，如果两个应用程序都在运行并且您增加了代理上的分区，则空闲的消费者将参与重新平衡。