如何在 CQRS 中处理读取模型答案

【问题标题】：How to process Read Model in CQRS如何在 CQRS 中处理读取模型
【发布时间】：2019-02-16 22:41:22
【问题描述】：

我们想在我们的新设计中实现 cqrs。我们对处理命令处理程序和读取模型有一些疑问。我们了解到，在处理命令时，我们应该对 aggregateId 采取乐观锁定。但是在处理 readModels 时应该考虑什么方法。我们是否应该锁定整个 readModel 或 aggregateId 或者在处理读取模型时从不锁定。

案例 1. 锁定整个 readmodel 时 -> 最安全，但速度不好。

案例 2 - 锁定 aggregateId。这里可能会出现两个问题。如果我们明智地使用 lock aggregateId -> 那么如果读取模型服务器重新启动会怎样。它不知道从哪里重新开始。

案例 3 - 永远不要锁定。在这种方法中，我认为数据可能处于损坏状态。例如，生成了一个订单插入事件，并且通过一些工作流/传奇，订单更新事件也发生了。如果订单更新事件先出现而订单插入事件尚未处理怎么办？

希望我能够解决我的问题。

【问题讨论】：

这有关系吗？使用 CQRS 时，您通常会在最终一致性的假设下工作。您通常将事件泵入队列，以保证您执行另一个工作人员的顺序，从队列中获取项目并更新读取存储。如果在一个请求中订单似乎尚未关闭更新或没有更新，它通常不是那么糟糕，因为读取存储只是提供信息。您的权威存储是写入存储（用于通过重播事件来填充您的聚合）
readStore 不仅仅是提供信息，我认为您所有的 UI 数据都来自 ReadModel 而不是来自聚合（写入模型）。假设我们有多个工作人员正在处理事件总线。一名工人已弹出下订单事件。正在处理并且意味着当另一个工作人员弹出 orderUpdated 事件时。现在如果第一个工人实际上还没有完成订单插入，就会出现问题。
我们正在通过总线或流序列化读取模型更新。如果您有一个从流中读取事件并应用于读取模型的工作进程 - 您不需要锁。

标签： domain-driven-design cqrs wolkenkit

【解决方案1】：

如果您不在 Readmodel 中同时处理事件，则不需要锁。当您有单个 Readmodel 实例（可能在微服务中）时就是这种情况，它轮询事件并按顺序处理它们。

如果您有一个同步的 Readmodel（即与 Writemodel/Aggregate 在同一进程中），那么您很可能需要锁定。

要记住的重要一点是，Readmodel 很可能与 Writemodel 不同。可能有很多 Writemodel 类型的事件被投影在同一个 Readmodel 中。例如，在电子商务商店中，您可以有一个 ListOfProducts，它从 Vendor 和 Product Aggregates 投射事件。这意味着，当我们谈论 Readmodel 时，我们不能简单地指“聚合”，因为不涉及单个聚合。在电子商务的情况下，当我们说“聚合”时，我们可能指的是Product 聚合或Vendor 聚合。

但是要锁定什么？这里取决于数据库技术。您应该锁定可以锁定的最小受影响读取实体或集合。在包含产品列表（读取实体，而不是聚合！）的 Readmodel 中，当事件仅影响一个产品时，您应该只锁定该产品（即ProductTitleRenamed）。

如果某个事件影响了更多产品，那么您应该锁定整个集合。例如，VendorWasBlocked 会影响所有产品（它应该从该供应商处删除所有产品）。

如果您想从离开的地方重试/恢复，您需要锁定具有非幂等副作用的事件，以防 Readmodel 的更新程序在事件处理期间失败。如果事件具有幂等副作用，则可以安全地重试。

为了知道在 Readmodel 失败的情况下从哪里恢复，您可以在 Readmodel 中存储最后处理的事件的序列。在这种情况下，如果实体更新成功，则最后处理的事件的序列也会被保存。如果它失败了，那么你就知道这个事件没有被处理。

【讨论】：

感谢康斯坦丁。假设我们发生了两个下订单事件。现在在读取模型中，两个实体都是独立的。但是当 readModel 处理事件时，如果第二个订单放置事件，等到第一个放置事件完全处理完毕，否则读取模型无法保持其状态，如果我们需要重新启动服务器，它已经处理了哪个事件。我们实际上正在研究异步的nodejs。
@RohitBansal 视情况而定。如果您序列化事件处理（所以如果您等待），那么您必须只跟踪最后处理的事件。如果您不序列化并并行执行，那么您应该存储您已处理的所有事件 ID。我看不到如何安全地在并行事件中执行，因为您需要按照它们生成的顺序处理来自聚合的事件。否则，您在 ProductCreated(id:123) 之前处理 ProductTitleUpdated(id:123)
@RohitBansal 换句话说，您应该至少对每个聚合流进行序列化/等待
这就是我要问的。如果串行处理 - 那么只需要跟踪最后一个事件 ID，但在这种情况下，我们可能无法分配负载。我说的对吗？
@RohitBansal 你可以。我对此给出了答案。我搜索它给你看。

【解决方案2】：

例如，生成了一个订单插入事件，并且通过一些工作流/传奇，订单更新事件也发生了。如果订单更新事件先出现而订单插入事件尚未处理怎么办？

如果您考虑它们轮询有序的事件序列，而不是对无序的通知做出反应，阅读模型通常更容易推理。

单个读取模型可能依赖于来自多个聚合的事件，因此聚合锁定不太可能是您最普遍的答案。

这也意味着，如果我们进行轮询，我们需要跟踪多个数据流的位置。换句话说，我们的读取模型可能包含元数据，这些元数据告诉我们使用了每个源的版本。

锁定可能取决于您的后备存储/缓存的性质。但乐观的态度

读取当前表示
计算新的表示
比较和交换

同样，通常很容易推理。

【讨论】：

谢谢。假设我们发生了两个下订单事件。现在在读取模型中，两个实体都是独立的。但是当 readModel 处理事件时，如果第二个订单放置事件，等到第一个放置事件完全处理完毕，否则读取模型无法保持其状态，如果我们需要重新启动服务器，它已经处理了哪个事件。我们实际上正在研究异步的nodejs。