事件溯源/CQRS 读取模型 - 预测

Posted 2023-03-08

【中文标题】事件溯源/CQRS 读取模型 - 预测

【英文标题】：Event sourcing / CQRS read model - projections

【发布时间】：2018-04-28 22:39:14

【问题描述】：

我有一个在 AWS Lambda 上运行的基于微服务的应用程序。其中两个最关键的微服务使用事件溯源/cqrs。

背景：（这也是我整理思路）

我正在使用 this library 并将事件存储在 DynamoDB 中，并将预测存储在 AWS S3 中。

写入部分的工作原理很像：每个命令调用都会从 DynamoDB 加载聚合的当前状态（通过处理程序运行事件和/或加载缓存的聚合），它会根据某些情况决定接受或拒绝命令业务逻辑，然后使用 KeyConditionExpression: 'aggregateId = :a AND version >= :v' 写入 DynamoDB，其中版本是为该聚合处理的事件计数。如果存在冲突，则写入失败。对我来说似乎是一个很好的系统！

然后将每个事件广播到 SNS（主题名称是服务名称），以便其他服务可以根据需要对事件做出反应。

我真正挣扎的部分是阅读。投影存储在 S3 中，并使用为每个事件源处理的最后一个 commitId 进行标记。当读取查询进入时，它会从 S3 （对于所有聚合） 加载整个投影状态，查询所有较新事件的事件源，计算最新状态（同样，对于所有聚合 - 并写入S3 更新的对象），并根据查询参数返回状态的相关部分。

我的问题：（或其中之一）

我认为我做错了预测。

我的大多数预测仅按重要属性对 id 进行分组，因此文件保持相对较小。但我还需要一种方法来检索单个聚合。为此使用投影似乎很疯狂，因为我需要每次加载整个状态（即每个投影聚合）应用新事件，然后检索我想要的记录（它甚至可能没有改变）。

这就是我现在正在做的事情，它的表现很好（

另一个问题是查询。我需要为我需要查询的每个属性建立一个投影映射值来匹配聚合ID！一定有更好的方法！

无论我如何看待这个问题，预测总是需要整个当前状态 + 任何新事件才能返回甚至没有更改的单个记录。

【参考方案1】：

我认为我做错了预测。

我也这么认为；听起来您的查询与您的预测相结合

当读取查询进入时，它会从 S3 加载整个投影状态（针对所有聚合），查询所有较新事件的事件源，计算最新状态

是的，这听起来像一团糟。或者更具体地说，这听起来像是查询正在触发投影要完成的工作。

如果您可以将查询与投影分离，那么事情会变得更容易。基本思想是您的查询不描述当前状态，它们描述的是上次投影运行时的状态。

相同的想法，不同的拼写：您从缓存在 S3 中的文档中回答查询。当检测到新事件时，您的投影会运行，根据需要加载新数据，计算新文档并替换缓存中的条目。

我想到了一个三角形

命令将信息从外部带到记录簿中 投影将信息从记录簿带入缓存 查询将信息从缓存带到外部世界

三角形的每条边都与其他边异步运行。

我建议您从查询开始 - 您需要哪些文档来支持每个查询？您必须达到的延迟目标是什么？然后开始权衡取舍 - 对于这个新查询，我是从现有文档创建结果，还是需要使用更细粒度构建的新文档？

如果我理解正确，我应该在事件进入时触发投影更新，而不是在进行查询时聚合。这使我不必在每次查询时都在事件存储中查询新事件

是的，而且……事件只是一种触发方式；您还可以通过时钟触发投影过程（每 15 分钟检查一次以查看我们是否需要更新）或人工操作员的突发奇想（嗯，看起来您的帐户余额已过时，让我尝试更新一下为你）。不止一种方法可以做到这一点，您可以混合搭配策略。

我仍然需要加载整个状态，无论是在更新投影时还是在加载单个聚合时。

不一定。没有规则说您不能使用以前缓存的表示作为起点，然后从记录簿中仅提取您需要的更改。

例如，假设您正在构建一个组合聚合 Aid:7 和 Bid:9 的视图。你抓住缓存的副本，查看它的元数据（你把它放在你以前写的地方），然后在里面找到像metadata:A:id:7, version:21, B:id:9, version:19 这样的东西。现在您只需要加载上次使用的事件之后的事件，更新内存中的本地副本，更新元数据的本地副本，并将批次推送到缓存。

【讨论】：

非常感谢您的回答！因此，如果我理解正确，我应该在事件发生时触发投影更新，而不是在进行查询时聚合。这使我不必在每个查询中查询事件存储中的新事件，但我仍然需要加载整个状态，无论是在更新投影时还是在加载单个聚合时。正确的？还是我错过了什么。

【参考方案2】：

我不熟悉你们的技术基础设施，但我实施预测的方式如下：

每个域事件都有一个跨越所有聚合根的全局序列号。投影是具有任意名称和由该全局序列号表示的最后处理位置的读取模型。我可以随时添加一个新投影及其事件处理程序，它将从位置 0 开始。我可以随时清除投影并将位置重新设置为 0。我还可以使用添加新投影的组合将替换现有的投影，即使需要几天的时间也可以构建，然后删除旧的。

有一项服务可以监视投影并几乎像队列一样使用事件存储。投影服务在当前位置之后检查具有全局 ID 的事件，并将这些事件交给处理程序，然后更新位置。这是您的投影甚至可以过滤事件类型以提高性能的地方。

这是基本思想。你的预测就是你所查询的。一旦投影赶上事件存储的“头部”，来自事件存储的事件将被涓流馈送到投影中。

我不太确定这将如何转化为您的技术领域。我有一个名为 Shuttle.Recall 的实验正在进行 C# 如果你想看看以获得一些想法。