在处理来自 Kafka 的消息时避免数据丢失

Question

寻找设计我的Kafka消费者的最佳方法。基本上，我想看看什么是避免数据丢失的最佳方法，以防在处理消息期间出现任何异常/错误。

我的用例如下。

a）我使用SERVICE来处理消息的原因是 - 将来我计划编写一个ERROR处理器应用程序，该应用程序将在一天结束时运行，它将尝试再次处理失败的消息（不是所有消息，而是由于任何依赖项（如父级缺失）而失败的消息）。

b）我想确保没有消息丢失，所以我会将消息保存到一个文件中，以防在将消息保存到DB时出现任何问题。

c)在生产环境中，可能有多个消费者和服务实例在运行，因此多个应用程序尝试写入同一个文件的可能性很高。

Q-1）写入文件是避免数据丢失的唯一选择吗？

Q-2）如果这是唯一的选择，如何确保多个应用程序写入同一文件并同时读取？请考虑将来一旦构建错误处理器，它可能正在读取同一文件中的消息，而另一个应用程序正在尝试写入该文件。

ERROR PROCESSOR-我们的源遵循事件驱动的机制，并且很有可能有时依赖事件（例如，某物的父实体）可能会延迟几天。因此，在这种情况下，我希望我的ERROR PROCESSOR能够多次处理相同的消息。

Answer 1

如果在写入数据库之前不提交消耗的消息，那么在Kafka保留消息时不会丢失任何内容。这样做的折衷是，如果消费者确实提交了数据库，但Kafka偏移提交失败或超时，您将再次消费记录，并且可能在您的服务中处理重复数据。

即使你写了一个文件，你也不能保证排序，除非你打开每个分区的文件，并确保所有消费者只在一台机器上运行（因为你在那里保留状态，这不是容错的）。重复数据删除也需要处理。

此外，您可以查看Kafka Connect框架，而不是将自己的消费者写入数据库。为了验证消息，您可以类似地部署Kafka Streams应用程序，将输入主题中的不良消息过滤到发送到数据库的主题中

Answer 2

我以前遇到过类似的事情。所以，直接进入你的问题:

> < li>

不一定，您也许可以在一个新的主题(比如- error-topic)中将这些消息发送回Kafka。因此，当您的错误处理器准备就绪时，它可以监听this error-topic并在这些消息到来时使用它们。

我认为这个问题已经在对第一个问题的答复中得到了解决。因此，Kafka可能是一个更好的选择，因为它是为解决这类问题而设计的，而不是使用一个文件来同时读写和打开多个文件句柄。

注意：以下几点只是基于我对你的问题领域的有限理解，值得思考的。所以，你可以选择安全地忽略这一点。

关于< code>service组件的设计，还有一点值得考虑——您还可以考虑通过将所有错误消息发送回Kafka来合并第4点和第5点。这将使您能够以一致的方式处理所有错误消息，而不是将一些消息放在error DB中，一些放在Kafka中。

编辑：基于关于ERROR PROCESSOR要求的附加信息，这里是解决方案设计的图解表示。

我故意将ERROR PROCESSOR的输出保持为抽象，只是为了保持它的通用性。

希望这有帮助！