如何在消息驱动的微服务系统中处理基于多个事件的副作用？答案

【问题标题】：How to handle side effects based on multiple events in a message driven microservice system?如何在消息驱动的微服务系统中处理基于多个事件的副作用？
【发布时间】：2018-03-16 19:25:29
【问题描述】：

我们目前在消息驱动的微服务环境中工作，我们的一些消息/事件是事件源的（使用 Apache Kafka）。现在我们正在努力实现更复杂的业务需求，如果我们必须考虑多个事件来创建新的事件和副作用。

在当前情况下，我们正在使用可能产生错误的设备，并且我们已经对其进行了处理，并且有一个包含 ERROR_OCCURRED 和 ERROR_RESOLVED 事件的主题（因此它们是有序的）。我们还确保有关特定设备的所有消息始终进入同一分区。两条消息共享一个标识该特定错误事件的 ID。我们已经有了一个使用这些事件并为我们的客户提供 API 的投影，s.t.他们可以看到所有发生的错误及其当前状态。

现在我们必须处理以下要求：

报告错误

我们需要一个推送系统，向我们的外部合作伙伴报告设备错误，但仅在 15 分钟后并且在该时间段内未解决。我们的第一种方法是使用所有 ERROR_RESOLVED 事件，存储 ID 并让另一个消费者以延迟的方式处理 ERROR_OCCURRED 事件（例如，如果它的时间戳至少 15 分钟旧，则仅使用该主题的下一个 ERROR_OCCURRED 事件）。然后，我们将能够知道该特定错误是否已经解决并且不需要报告（因为它们与相应的 ERROR_RESOLVED 事件共享一个公共 ID）。否则，我们向外部合作伙伴发送 HTTP 请求，并在新主题上创建 ERROR_REPORTED 事件。 有没有更好的延迟和条件消息处理方法？

我们还必须考虑以下特殊用例：

服务重新启动：目前我们计划将已解决的错误列表保留在内存中，因此如果服务重新启动，则必须从头开始创建该列表。我们可以只重播 ERROR_RESOLVED 消息，但这可能需要一些时间，在此期间不应处理任何 ERROR_OCCURRED 事件，因为这可能会导致报告在不到 15 分钟内解决的错误，但我们只是没有意识到这一点。 在回放与“正常”处理方面有什么好的做法吗？
缩放：我们可能随时增加或减少服务实例的数量，因此分区分配可能会在运行时发生变化。如果我们在使用 ERROR_RESOLVED 事件时为每个服务实例创建一个消费者组，那应该不是问题，s.t.每个实例都知道所有已解决的错误，同时仍然只处理其分配的分区的 ERROR_OCCURRED 事件（在另一个由所有实例共享的消费者组中）。 是否有更好的方法来处理分区重新分配和内部状态？

提前致谢！

【问题讨论】：

寻求调试帮助的问题（“为什么这段代码不起作用？”）必须包括所需的行为、特定的问题或错误以及重现它所需的最短代码在问题本身。没有清晰的问题陈述的问题对其他读者没有用处。请参阅：How to create a Minimal, Complete, and Verifiable example。

标签： events apache-kafka state event-sourcing

【解决方案1】：

对于副作用，我会在事件存储中记录所有“侧面”动作。在您的特定示例中，当需要发送通知时，我会调用 SEND_NOTIFICATION 命令来发出 NOTIFICATION_SENT 事件。这些事件将由一些执行实际 HTTP 请求的工作进程处理。

实际上，我会更详细地说明这一点，因为通知可能会失败，所以我会有两个事件 NOTIFICATION_REQUIRED 和 NORIFICATION_SENT，所以我们可以重试失败的通知。

最后你的逻辑是“如果错误没有在 15 分钟内解决并且通知没有发送 - 发送通知（或者如果它错过了它的时间框架就丢弃）”

【讨论】：