在 Apache Storm 中处理 kafka 消息时如何确保只使用一次语义

Question

我只需要在我的应用中交付一次。我探索了 kafka 并意识到要只生成一次消息，我必须在生产者配置中设置 idempotence=true。这也设置了acks=all，使生产者重新发送消息，直到所有副本都提交它。为了保证消费者不做重复处理或留下任何未处理的消息，建议在同一个数据库事务中将处理输出和偏移量提交到外部数据库，这样要么都被持久化，要么都不持久，避免重复和不处理。

在消费者中，如果消费者首先提交消息但在处理之前失败，则消息将被处理；如果消费者首先处理消息但在提交之前失败，则消息被处理多次。

Q1. 现在我在猜测如何使用 Apache Storm 进行模仿。我想通过在KafkaBolt 中设置idemptence=true 可以确保生成消息。我说的对吗？

我在猜测如何确保在 Storm 中处理丢失和重复的消息。例如，this doc page 表示如果我锚定一个元组（通过将其作为第一个参数传递给OutputCollector.emit()）然后将元组传递给OutputCollector.ack() 或OutputCollector.fail()，Storm 将确保数据丢失。这正是它所说的：

现在您已经了解了可靠性算法，让我们回顾一下所有故障案例，看看 Storm 在每种情况下如何避免数据丢失：

• 由于任务死亡，元组未得到确认：在这种情况下，失败元组的树根的 spout 元组 id 将超时并被重放。

• Acker 任务终止：在这种情况下，acker 跟踪的所有 spout 元组都将超时并被重放。

• Spout 任务终止：在这种情况下，spout 与之对话的源负责重播消息。例如，像 Kestrel 和 RabbitMQ 这样的队列会在客户端断开连接时将所有未决消息放回队列中。

Q2. 我想这可以确保消息不会未被处理，但不会避免重复处理消息。我对此是否正确？此外，Storm 还提供了其他任何东西来确保我缺少像 kafka 这样的语义吗？

Answer 1

关于 Q1：是的，您可以通过设置该属性从 KafkaBolt 获得相同的行为，KafkaBolt 只需包装 KafkaProducer。

关于消费端的语义，Storm 的选择与 Kafka 的选择相同。当您从 Kafka 读取消息时，您可以选择在处理之前或之后提交（例如写入数据库）。如果您之前这样做，并且程序崩溃，您将丢失消息。我们称之为at-most-once processing。如果您在之后执行此操作，则如果程序在处理之后但在提交之前崩溃（称为at-least-once processing），您将面临处理同一消息两次的风险。

所以，关于 Q2：是的，使用锚定元组和 acking 将为您提供at-least-once 语义。不使用锚定元组会给你at-most-once。

是的，Storm 还提供了其他东西来确保精确一次的语义，称为 Trident，但它要求您以不同的方式编写拓扑，并且您的数据存储必须适应它，这样才能进行消息重复数据删除。请参阅https://storm.apache.org/releases/2.0.0/Trident-tutorial.html 的文档。

另外提醒您：当 Storm（或 Kafka）的文档谈论完全一次性语义时，对您将进行何种处理做出了一些假设。例如，当 Storm 的 Trident 文档谈论完全一次时，假设您将调整您的数据库，以便您可以决定何时给出消息是否已经存储。当 Kafka 的文档谈到恰好一次时，假设您的处理将从 Kafka 读取，进行一些计算（很可能没有副作用）并写回 Kafka。

这只是说对于某些类型的处理，您可能仍需要在at-least-once 和at-most-once 之间进行选择。如果可以让处理幂等，at-least-once 是一个不错的选择。

最后，如果您的处理符合“从 Kafka 读取、进行计算、写入 Kafka”模型，那么您可能会从 Kafka Streams 中获得比 Storm 更好的语义，因为 Storm 无法提供 Kafka 可以提供的完全一次性语义在那种情况下。