如何在 CQRS / 事件溯源中以确定的方式重播？答案

【问题标题】：How to replay in a deterministic way in CQRS / event-sourcing?如何在 CQRS / 事件溯源中以确定的方式重播？
【发布时间】：2020-02-04 04:44:47
【问题描述】：

在基于 CQRS / ES 的系统中，您将事件存储在事件存储中。这些事件引用一个聚合，并且它们相对于它们所属的聚合具有顺序。此外，聚合是一致性/事务边界，这意味着任何事务保证仅在每个聚合级别上给出。

现在，假设我有一个读取模型，它使用来自多个聚合的事件（这很好，AFAIK）。为了能够以确定的方式重放读取模型，事件需要某种全局排序，跨聚合 - 否则您将不知道是在 B 之前还是之后重放聚合 A 的事件，或者如何混合他们。

实现此目的的最简单解决方案是在事件上使用时间戳，但通常时间戳不够精细（或者，换句话说，并非所有数据库都是平等的）。另一种选择是使用全局序列，但这在性能方面很差，并且会阻碍扩展。

你如何解决这个问题？或者是我的基本假设，即读取模型的重放应该是确定性的，错了吗？

【问题讨论】：

你能举一个例子，说明什么时候聚合 A 在聚合 B 之前重播很重要？
假设您有两个聚合，group 和 user，并且用户可以是组的一部分。当然，您不希望组和用户成为同一聚合的一部分，因为这会大大减少软件的并行使用。但是，在用户能够加入组之前，该组必须存在。虽然这可以在写入端轻松解决，但如何决定在读取端以何种顺序重播组和用户的事件，以避免在groups of each user 视图中尝试将组添加到用户记录中用户还没到？
@GoloRoden（删除了我之前的评论，因为我想我没理解） - 在我们自己的实现中，我们在事件存储中使用了流名称，并且能够为一组特定的流重放事件，从而允许我们控制命令。即： $ce-groups 然后 $ce-users 等...
@Nope 好的，但是你是自动计算出这个顺序，还是开发人员必须定义的东西？

标签： cqrs event-sourcing

【解决方案1】：

我看到了这些选项：

全局序列
- 如果你的数据库允许，你可以使用timestamp+aggregateId+aggregateVersion作为索引。这在分布式数据库的情况下通常效果不佳。
- 在分布式数据库中，您可以使用vector clock 获取全局序列而无需锁定。
每个读取模型内的事件序列。您可以将所有事件存储在读取模型中，并在应用投影函数之前根据需要对它们进行排序。
允许不确定性并处理它。例如，在您的示例中，如果 add_user 事件到达时没有组 - 只需为读取模型创建一个空组记录并添加一个用户。当 create_group 事件到达时 - 更新该组记录。毕竟，您已经签入了那里的 UI 和/或命令处理程序是一个有这个aggregateId的组，对吧？

【讨论】：

if there is no group when add_user event arrives - just create a group and add a user - 如果 AddUser 命令处理程序检查组存在，如果组不存在，命令不会被拒绝吗？您是否建议如果该组不存在 AddUser 命令处理程序应该发送一个 AddGroup 命令？如果为无效组触发 AddUser 命令，这不会引入紧密耦合并可能引入错误吗？
@Nope - 我们正在谈论读取端，所以我假设已检查业务规则 - 意味着事件在流中是正确的，只是它们的顺序不正确。所以我假设用户已添加到现有组中，否则 UI 或命令处理程序将不允许此事件存在。
是的，否则该事件将不存在。我仍然对在您的建议中创建组的假设感到困惑，因为 UserAdded 事件只会将用户数据写入读取数据存储，并且不必担心该组在该点和查询/读取端存在，查询将无法找到组的用户，该组在重播组事件之前不存在。对不起，我显然只是误解了一些事情。我自己还在琢磨这些事情:)
根据我们的经验，我们可以看到一些事件以毫秒为单位延迟。因此，在此示例中，读取模型中的组记录将在一秒钟内完全更新 - 这通常是可以接受的。在我们的系统 (reSolve) 中，我们使用这里提到的矢量时钟，所以我们没有这个问题。但是矢量时钟不是免费的——如果顺序错误，我们需要重新查询事件，所以如果可能的话，忽略顺序问题是最便宜的解决方案。

【解决方案2】：

你如何解决这个问题？

这是已知问题，当然，简单的时间戳、全局序列、事件幼稚方法也无济于事。
使用带有弱时间戳的vector clock 来枚举您的事件并使用矢量光标来读取它们。这保证了一些稳定的确定性顺序来混合聚合之间的事件。即使每个线程都有时钟同步间隙，这也可以工作，这是数据库集群的常规用例，因为完美的时间戳同步是不可能的。
这也自动为以后从事件存储和事件总线无缝混合读取事件提供了可能性，并排除了不同聚合事件之间的任何数据库锁。

算法草案：
1）确定数据库中同时交易的实际数量，例如集群中的最大工作人员数。
由于每个事件仅在一个线程中写入一个事务，您可以将其唯一 id 确定为元组(thread number, thread counter)，其中线程计数器是当前线程上处理的事务量。
计算事件弱时间戳为MAX(thread timestamp, aggregate timestamp)，其中聚合时间戳是当前聚合的最后一个事件的时间戳。

2) 为通过线程号边界读取事件准备矢量光标。从每个线程按顺序读取事件，直到时间戳间隙超过允许值。允许的弱时间戳差距是事件读取性能和保留本机事件顺序之间的交易。
最小值是集群线程同步时间增量，因此事件以本机聚合混合顺序到达。最大值为无穷大，因此事件将被聚合吐出。当使用像 postgres 这样的 RDBMS 时，可以通过智能 SQL 查询自动确定该值。

您可以看到 saving events 和 loading events 的 PostgreSQL 数据库的参考实现。对于 4GB RAM RDS Postgres 集群，保存事件性能约为每秒 10000 个事件。

【讨论】：