高负荷下的Kafka平衡问题

Question

使用kafka版本2.11-0.11.0.3发布10,000条消息（所有消息的总大小为10MB)，将有2个消费者（具有相同的group-id）作为并行处理来消费该消息。在消费过程中，两个消费者都消费了相同的消息。

以下错误/警告是Kafka抛出的

警告：此成员将离开组，因为使用者轮询超时已过期。这意味着对poll（）的后续调用之间的时间长于配置的max.poll.interval.ms，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加max.poll.interval.ms或通过减少poll（）中返回的批的最大大小和max.poll.records来解决这一问题。

信息：尝试心跳失败，因为组正在重新平衡

信息：向协调器发送LeaveGroup请求

警告：同步自动提交偏移量{INGEST-DATA-1=OFFSETANDMETADATA{OFFSET=5506,LeaderEPOCH=NULL,METADATA=“”}}失败：无法完成提交,因为组已重新平衡并将分区分配给其他成员.这意味着对poll（）的后续调用之间的时间比配置的max.poll.interval.ms长，这通常意味着poll循环花费了太多时间处理消息。您可以通过增加max.poll.interval.ms或通过减少poll（）中返回的批的最大大小和max.poll.records来解决这一问题。

向kafka提供了以下配置

server.properties

max.poll.interval.ms=30000
group.initial.rebalance.delay.ms=0
group.max.session.timeout.ms=120000
group.min.session.timeout.ms=6000

Consumer.Properties

session.timeout.ms=30000 
request.timeout.ms=40000

应该改变什么来解决多次消费？

Answer 1

你的消费者是同一群吗？如果是，如果使用者离开/死亡/超时而没有提交它已处理的一些消息，那么您将有多个使用者。

如果您的所有消息都被两个消费者使用，那么您可能没有为他们设置相同的组id。

更多信息：

所以您已经为所有消费者设置了相同的组id，很好。您所处的情况是，集群/代理认为某个消费者已经死亡，因此将负载重新平衡到另一个消费者。另一个将从上次提交的地方开始使用。

因此假设使用者C_A从分区P_1读取最多为100的偏移量，然后处理它们，然后提交'100'，然后读取最多为200的偏移量，然后处理它们，但不能提交，因为代理认为C_A已死。

代理将分区P_1重新分配给使用者C_B，使用者C_B将从组的最后一次提交开始（组的最后一次提交为100)，读取最多200，处理并提交200。

所以你的问题是如何避免消费者被认为是死了（我假设它不是死的）？

答案已经在您的问题中的黄色警告消息中：您可以告诉您的使用者在一次轮询中消耗更少的消息(max.poll.records)，以减少两次轮询之间对代理的处理时间，和/或您可以增加max.poll.interval.ms，告诉代理在认为您的使用者已死亡之前等待更长时间...