基于Kafka的事件溯源并发写入

Question

我一直在考虑在事件溯源配置中使用Apache Kafka作为事件存储。发布的事件将与特定资源相关联，交付到与资源类型相关联的主题，并通过资源ID将其分片到分区中。例如，创建资源类型为Folder且ID为1的资源将生成一个FolderCreate事件，该事件将被交付到“folders”主题中，在该主题的分区中根据ID 1对总分区数进行划分。即使我不知道如何处理使日志不一致的并发事件。

最简单的情况是存在两个可以使彼此无效的并发操作，例如更新文件夹和销毁同一文件夹的操作。在这种情况下，该主题的分区可能包含无效序列[FolderDestroy，FolderUpdate]。该情况通常通过对事件进行版本控制（正向并发性和幂等性）解决，具体请参见这里 ，但Kafka不支持该功能。

那么，有什么方法可以确保在这些情况下Kafka日志本身的一致性呢？

- Jesuspc

---

你读过这个吗？https://dev59.com/o2Mm5IYBdhLWcg3wT9qU - Constantin Galbenu

Answer 1

我认为可以使用Kafka对聚合（在DDD意义下）或“资源”进行事件溯源。以下是一些注意事项：

1. 按分区序列化写入，每个分区使用一个进程（或多个分区）来管理此过程。确保将消息串行地发送到同一Kafka连接，并在向命令发送方报告成功之前使用ack=all，如果您无法承担回滚成本。确保生产者进程跟踪每个资源的当前成功事件偏移/版本，以便在发送消息之前自己进行乐观检查。
2. 由于即使写入实际上成功了，也可能返回写入失败，因此您需要重试写入并通过在每个事件中包含一个ID来处理去重复，或通过重新读取流（最近的消息）来重新初始化生产者，以查看写入是否实际起作用。
3. 原子地写入多个事件-只需发布包含事件列表的组合事件。
4. 按资源ID查找。这可以通过在启动时从分区读取所有事件（或从特定的跨资源快照读取所有事件），并将当前状态存储在RAM中或缓存在数据库中来实现。

https://issues.apache.org/jira/browse/KAFKA-2260 可以更简单地解决1，但似乎已经停滞不前。

Kafka Streams似乎为您提供了很多功能。例如，4是一个KTable，您可以让事件生产者使用它来确定事件是否对当前资源状态有效，然后再发送它。

- TomW

---

如果每个分区只有一个进程，您无需等待确认 - 所有处理都可以在该进程内串行化，对于排序/一致性而言，某些消息仍然在传输中并不重要（拥有保证写入是另一个问题，可能不应同时混合解释）。 - Artur Biesiadowski

但是如果资源1，事件写入A失败，然后您在不等待A的确认的情况下编写事件B，而B编写成功呢？因此，B在没有A的情况下进入流中，破坏了不变量。或者A被延迟了，最终在B之后进入流中。或者这根本不可能，因为两者将使用相同的TCP连接？无论如何，如果是这种情况，这会使事情变得简单一些。 - TomW

我认为不可能从单个程序发送事件A和B到单个kafka通道，最终只有B存在。它将是什么都没有、仅A或A和B。生产者连接是无损TCP流，具有重新连接功能。当然，如果你故意搞砸了kafka代理（将消息删除时间设置为几秒钟，在处理过程中重新启动并清除内容等），一切皆有可能，但在正常操作（即使有错误）中，事件不应该从生产者流的中间丢失。 - Artur Biesiadowski

很遗憾这种情况是有可能出现的，我们也见过。生产者发送A，但代理却拒绝了它，因为重新平衡已经发生了，并且它不再是那个分区的Leader。生产者发现新的Leader并发送B。B被存储，但A没有。只有在您更注重吞吐量而非一致性并且愿意承担（相对）正常操作下消息丢失的代价时才使用acks=0。 - Michal Borowiecki

对于＃2，我强烈建议不要尝试读取流的内容以确定记录是否已经发布，然后再次发布，以避免重复。这会显着增加生产者实现的复杂性。 Kafka背后的一般设计哲学是至少一次交付，意味着消费者生态系统中具有幂等行为。您的消费者应该已经能够处理重复事件。 - Emil Koutanov

我应该更新答案 - Kafka现在支持幂等写入，所以＃2不再是问题（如果您使用事务API，则可以重试而不会获得重复）。 - TomW