避免在Apache Kafka Consumer中重复消息的有效策略

Question

我已经学习apache
kafka一个月了。但是，我现在陷入了困境。我的用例是，我有两个或多个在不同计算机上运行的使用者进程。我进行了一些测试，在其中我在kafka服务器中发布了10,000条消息。然后，在处理这些消息时，我杀死了一个使用者进程并重新启动了它。消费者正在将处理后的消息写入文件中。因此，使用结束后，文件显示了超过1万条消息。因此，某些消息是重复的。

在使用者过程中，我已禁用自动提交。消费者手动批量提交偏移量。因此，例如，如果将100条消息写入文件，使用者将提交偏移量。当单个使用者进程正在运行并且崩溃并恢复时，可以通过这种方式避免重复。但是，当一个以上的使用者运行时，其中一个崩溃并恢复时，它将重复的消息写入文件。

有没有有效的策略来避免这些重复的消息？

Answer 1

最简洁的答案是不。

您要查找的是一次精确的处理。尽管它似乎经常可行，但永远不要依赖它，因为总会有一些警告。

即使为了防止重复，您也需要使用简单的使用者。对于每个使用者，此方法的工作方式是：从某个分区使用一条消息时，将使用的消息的分区和偏移量写入磁盘。当使用者在故障后重新启动时，请从磁盘读取每个分区的上一个消耗的偏移量。

但是，即使采用这种模式，使用者也无法保证在失败后不会重新处理消息。如果使用者使用一条消息然后在将偏移量刷新到磁盘之前失败，该怎么办？如果在处理消息之前先写磁盘，如果在实际处理消息之前先写偏移量然后失败，该怎么办？即使您在每条消息之后将偏移量提交给ZooKeeper，也将存在相同的问题。

但是，在某些情况下，更精确的一次处理是可以实现的，但仅适用于某些用例。这仅要求将偏移量存储在与单元应用程序输出相同的位置。例如，如果编写一个对消息进行计数的使用者，则通过将最后计数的偏移量与每个计数一起存储，可以保证该偏移量与使用者的状态同时存储。当然，为了保证处理一次，这将要求您只消耗一条消息并为每条消息更新一次状态，这对于大多数Kafka消费者应用程序来说是完全不切实际的。从本质上来说，Kafka出于性能原因而批量使用消息。

通常，如果仅将其设计为幂等的，则您的时间将花费更多，并且应用程序将更加可靠。