apache-kafka - 多线程 Kafka Consumer 或 PerPartition-PerConsumer

Question

在实现 kafka 消费者时应该采取什么更好的方法。

目标从 Kafka 读取并写回 db。数百万行

方法一:
每个分区 - 每个消费者 - 等待消息消耗(即写回数据库)然后在轮询循环中继续下一个。

方法二:
每个分区 - 每个消费者 - 将记录发送到工作线程或线程池以写回数据库，然后提交偏移量并继续轮询。需要注意抵消管理。在这种情况下，不要等待消息写回数据库。继续轮询，将消息传递给工作线程。

关于他们两个的任何见解？

谢谢

Answer 1

方法一:
该方法仅适用于您可以估计消息处理时间的情况，否则不推荐使用。

问题:在这种方法中，主要问题是让消费者保持事件状态，如果您在再次调用 poll() 之前等待消息完全处理，则必须确保您的消费者在调用 poll() 之前应该处于事件状态，因为 kafka维护一个名为“session.timeout.ms”的属性。 kafka broker/cluster 对这个属性的值采取行动，如果消费者在“session.timeout.ms”的时间段内无法再次调用 poll()，broker 会标记消费者死亡，它会被踢出.现在，当消费者完成消息处理并再次调用 poll() 时，它被视为新加入者，并将再次像以前一样从偏移量开始提供记录集。牢记这种情况，消费者将陷入无限循环，永远不会继续其偏移量。

可能的解决方案 1:要使用这种方法，您需要具有以下副作用的以下属性“session.timeout.ms”的良好值(value):

1:值太低:消费者将被标记为死亡，如上所述，并且永远不会继续其偏移量，但是消息将被处理，但每次完成消息时，它将再次获得以前的消息+新消息。

2:值太高:Broker 检测消费者的真正故障会很晚，这将导致记录重复并影响整体吞吐量。

可能的解决方案 2:(仅对版本 0.10.1.x 有效) Kafka 在发布 (0.10.1.0) 中的官方修复。
在这种方法中，引入了两个值得注意的实体:一个新属性“max.poll.interval.ms”，它设置了客户端调用 poll() 之间的最大延迟，以及一个负责保持消费者事件的后台线程。因此，在某个场景中，当消费者调用方法 poll() 然后忙于消息处理时，内部后台线程将使心跳保持事件状态，因此消费者将保持事件状态。但是，这个内部后台线程本身将保持事件状态，直到属性“max.poll.interval.ms”的超时值保持有效。因此，该线程将等待消费者在“max.poll.interval.ms”的时间段内调用 poll() ，否则，它将发送离开请求并自行死亡。”

同样，这个解决方案中棘手的部分是找到这个属性的合适值:“max.poll.interval.ms”(非常重要，这个时间将是后台线程将保持心跳事件而不需要显式调用 poll())。

方法二:使用工作线程是一个好主意，但是您必须维护一个内部队列或对接收到的消息进行验证，这可能很复杂，而且您还需要针对自动提交使用手动提交。有关提交的更多信息，请参阅 this并搜索标题“提交和抵消”。

问题:在这种方法中，主要问题是跟踪收到的消息和成功处理的消息。因为，您的消费者将收到消息，它将消息传递给相应的工作线程，并将提交偏移量并继续接收更多消息。在此过程中，您必须注意以下问题:

如果收到消息并提交偏移量，但后来由于某种原因工作线程无法处理该消息，那么现在如何再次获取该消息呢？

如果消费者收到消息但没有空闲的工作线程要处理怎么办？

解决方案:可以有不同的方法来解决上述问题，一种方法是使用内部队列来保留消息和手动提交，只有当工作线程报告消息处理成功时才会发送这些消息和手动提交。然而，需要非常仔细的实现，因为它可能导致复杂的代码，也可能导致内存管理或线程问题。

建议:根据您的要求，您可以使用一种方法或另一种方法来解决上述可能的问题。但是我会推荐一个更强大的解决方案是使用分区暂停/恢复。以非常抽象的方式，您的消费者应该执行以下步骤:

1:poll() 用于消息。

2:暂停所有相应的主题/分区。

3:将消息分配给工作线程并等待它们的处理。

4:继续调用 poll() 但当分区暂停时，不会收到额外的消息，而消费者将保持事件状态。 (确保此时没有注册新主题)

5:如果所有工作线程都应报告消息处理成功/失败，则相应地提交偏移量。

6:恢复所有分区。

注意:根据您的场景和要求，可能有更好的方法或其他可能的解决方案。这只是一个想法或可能的解决方案之一。