Spring Cloud Stream Kafka - 如何实现幂等性以支持分布式事务管理（最终一致性）

Question

我有以下典型场景：

• 用于购买产品的订单服务。充当分布式事务的指挥官。
• 包含产品列表及其库存的产品服务。

• 一种支付服务。

      Orders DB               Products DB
         |                       |
  ---------------         ----------------          ----------------
  | OrderService  |       | ProductService |        | PaymentService |
   ---------------         ----------------          ----------------
         |                       |                         |
         |                --------------------             |
         --------------- | Kafka orders topic |-------------
                         ---------------------
  

正常流程是：

1. 用户订购产品。
2. 订单服务在 DB 中创建订单并在 Kafka 主题“订单”中发布消息以预留产品 (PRODUCT_RESERVE_REQUEST)。
3. 产品服务将其数据库中的产品库存减少一个单位，并在“订单”中发布一条消息说 PRODUCT_RESERVED
4. Order 服务获取 PRODUCT_RESERVED 消息并命令支付发布消息 PAYMENT_REQUESTED
5. 支付服务订购付款并回复一条消息已支付
6. 订单服务读取 PAYED 消息并将订单标记为 COMPLETED，完成交易。

我在处理错误情况时遇到了麻烦，例如：让我们假设：

5. 支付服务未能为产品收费，因此它发布消息PAYMENT_FAILED
6. 订单服务响应发布消息 UNDO_PRODUCT_RESERVATION
7. 产品服务增加数据库中的库存以取消预订并发布 PRODUCT_UNRESERVATION_COMPLETED
8. 订单服务完成交易，将订单的最终状态保存为 CANCELLED_PAYMENT_FAILED。

在这种情况下，假设无论出于何种原因，订单服务发布了 UNDO_PRODUCT_RESERVATION 消息但没有收到 PRODUCT_UNRESERVATION_COMPLETED 消息，因此它重新尝试发布另一个 UNDO_PRODUCT_RESERVATION 消息。

现在，假设同一订单的这两条 UNDO_PRODUCT_RESERVATION 消息最终到达 ProductService。如果我同时处理它们，我最终可能会为产品设置无效库存。

在这种情况下如何实现幂等性？

更新：

按照 Artem 的说明，我现在可以检测到重复的消息（通过检查消息标头）并忽略它们，但可能仍然存在以下情况，我不应该忽略重复的消息：

1. Order Service 发送 UNDO_PRODUCT_RESERVATION
2. 产品服务收到消息并开始处理它，但在更新库存之前崩溃。
3. Order Service 未收到响应，因此它重新尝试发送 UNDO_PRODUCT_RESERVATION
4. 产品服务知道这是一条重复的消息但是，在这种情况下它应该再次重复处理。

你能帮我想出一种方法来支持这种情况吗？我如何区分何时应该丢弃消息或重新处理它？

Answer 1

实际上，由于您提到通过 Apache Kafka 在多个微服务上组织事务的复杂性，我开发了另一个概念并写了一篇关于它的博客。

如果您陷入 Kafka 解决方案可能不再可行的复杂状态，您可能会觉得这是一本有趣的读物。这里解释的太长了，但基本上它使用了一个完全符合微服务原理的 J2EE 容器，并在 Spring Boot + Netflix 的帮助下在微服务之间提供了完整的事务支持。

Micro Services Fanout and Transaction Problems and Solutions with Spring Boot and Netflix

Answer 2

我们使用 spring-integration-kafka 在我们的微服务中使用 Kafka 生成和使用消息。在我们的例子中，我们将 org.springframework.messaging.Message 对象发送到主题，并在从字节数组反序列化后从主题中获取相同的类型。在消息实体中，除了消息有效负载之外，还有消息 ID、发送时间等标头值，这是您希望从一个微服务传输到其他微服务的实际对象。我们使用唯一的 message-id 值来实现幂等性。在生产者方面，您必须实现一些逻辑以确保消息在多次生产时的消息ID是相同的。这实际上与您的生产逻辑有关。在我们的例子中，我们使用使用本地事务发布事件，这在 Chris Richardson 的博客 https://www.nginx.com/blog/event-driven-data-management-microservices/ 中有很好的描述。使用这种方法，我们可以在生产者端使用相同的 message-id 重新创建 Message 对象。在消费者方面，我们将所有消费的消息 id 值保存到数据库中，并在处理接收到的消息之前检查此 id。如果我们看到一条 id 在我们的持久存储中的消息，我们只需忽略它。

在你的情况下，要实现幂等性：

• 您应该为消息保留一个唯一标识符，
• 在生产者方面，您必须在多次生产时生成相同的标识符，
• 在消费者端，你必须检查接收到的 id 来检测它之前是否被消费过

关于 UPDATE 中描述的第二种情况，

我认为你应该稍微改变主意。如果你想实现更适合微服务架构的发布-订阅机制，你不应该等待生产者端的响应。在这种情况下，您等待其他消息知道消费者是否消费了该消息，如果消费者没有消费，则再次发送。

下面的实现怎么样； 在生产者方面，您在生产者的事务中向 Kafka 发送消息。您应该在此处提供一种机制，以便仅提交生产者端的事务向 kafka 发送消息。这是原子性问题，我在上面给出了一个链接，显示了如何解决这个问题。

在 Consumer 端，你按顺序从 kafka 主题中轮询消息，只有当当前消息可以被消费时，你才能获得下一条消息。如果它没有被消费，你不应该得到下一条消息。因为下一条消息可能与当前消息相关，如果您使用下一条消息，您可能会破坏数据的一致性。当消息没有被消费时，它不是生产者关心的。在消费者方面，您应该提供重试和重播机制来消费消息。

我认为您不应该等待生产者方面的响应。 Kafka 是一个非常智能的工具，并且具有偏移提交功能，作为消费者，当您从主题轮询消息时，您不必消费消息。如果您在处理消息时遇到问题，您只需不提交偏移量来获取下一条消息。

通过上述实现，您不会遇到“我如何区分何时应该丢弃消息或重新处理它？”

问候...