2

我有以下典型场景：

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

  • 一种支付服务。

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

正常的流程是：

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

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

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

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

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

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

更新：

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

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

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

4

2 回答 2

2

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

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

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

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

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

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

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

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

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

问候...

于 2017-02-13T14:45:25.257 回答
-1

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

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

Spring Boot 和 Netflix 的微服务扇出和事务问题及解决方案

于 2017-05-16T11:24:58.847 回答