谁以及如何处理重播事件？

Question

我正在学习 DDD、CQRS 和事件溯源，但有些东西我无法弄清楚。命令触发聚合中的更改，一旦执行更改，就会触发一个事件。该事件随后由系统的其他部分处理并保存在事件存储中。但是，如果更改由命令触发，我不明白重播事件将如何重新创建聚合。

示例：如果我们有一家网上商店。 AddItemToCardCommand -> Card Aggregate 将项目添加到其卡片 -> ItemAddedToCardEvent -> 该事件由谁处理。 但是，如果重播该事件，则聚合不会将该项目添加到其卡片中。

总而言之，我的问题是我应该如何根据事件存储中的事件重新创建聚合？此外，任何关于如何以正确方式重播事件的一般建议都会受到重视。

Answer 1

为简单起见，我们假设一个无状态进程 - 我们的服务不会尝试将事物的副本保存在内存中，而是根据需要重新加载聚合。

服务接收AddItemToCardCommand:{card:123, ...}。我们在内存中没有card:123 的当前状态，所以我们需要创建它。我们通过从持久存储中加载card:123 的状态来做到这一点。因为我们选择使用事件源存储，所以我们从持久存储中读取的“状态”是服务先前写入的事件历史记录的表示。

事件历史包含您需要记住的所有信息，但不一定采用方便的“形状” - 仅附加列表是写入的绝佳数据结构，但不一定适合读取。

这通常意味着我们将“重播”事件以创建一个内存对象，然后我们可以使用它来回答有关我们接下来要编写的事件的问题。

回答简单查询时使用相同的模式：我们从存储中加载事件历史记录，将事件历史记录转换为更方便的形状，然后使用该形状来计算答案。

在查询延迟比及时性更重要的情况下，我们可能会设计查询处理程序来从缓存中读取方便的形状，而不是每次都尝试重新计算它们；并发运行的后台线程将负责定期唤醒以计算缓存的新内容。

使用异步进程从事件流中提取更新是一种常见模式； Greg Young 在他的Polyglot Data 演讲中讨论了这种方法的一些优势。

Answer 2

在理想的事件场景中，您的数据库中不会有一个已经构建好的聚合结构。通过运行到目前为止存储的所有事件，您反复到达最终数据结构。

让我用一些伪代码来说明将商品添加到购物车，然后获取购物车数据。

# Create a new cart
POST /cart/new

# Store a series of events related to the cart (in database as records, similar to array items)
POST /cart/add -> CartService.AddItem(item_data) -> ItemAddedToCart

一系列事件如下所示：

* ItemAddedToCart
* ItemAddedToCart
* ItemAddedToCart
* ItemRemovedFromCart
* ItemAddedToCart

当需要从数据库中获取购物车数据时，您构建一个新的购物车实例（或检索一个购物车实例，如果已持久化）并在其上重放事件。

cart = Cart(id=ID1)

# Fetch contents of Cart with id ID1
for each event in ID1 cart's events:
    if event is ItemAddedToCart:
        cart.add_item(event.data)
    else if event is ItemRemovedFromCart:
        cart.remove_item(event.data)

return cart

有时，当与购物车相关的事件过多时，您可能希望生成聚合结构并将其保存在数据库中。下一次，您可以从聚合结构保存点开始，并继续应用新事件。当要处理的事件过多时，这种优化有助于节省时间并提高性能。

Answer 3

可能有帮助的是不要将 command 视为更改状态，而是将 event 视为更改状态。事实上，我不太明白人们会怎么做。聚合中的命令处理程序将应用不变量，如果一切正常，将立即创建事件并调用一些应用它的方法 ([Apply|On|Do]MyEvent)。事后发生事件这一事实必然意味着系统的其他部分会处理它。然而，它是事件溯源所必需的。一旦你有了一个事件，你当然可以通过例如在服务总线上发布将它传递到系统的其他部分。

当您重放事件时，您调用的方法与命令调用的方法相同，以实际改变聚合的状态：

public MyEvent MyCommand(string data)
{
    if (string.IsNullOrWhiteSpace(data))
    {
        throw new ArgumentException($"Argument '{nameof(data)}' may not be empty.");
    }

    return On(new MyEvent
    {
        Data = data
    });
}

private MyEvent On(MyEvent myEvent)
{
    // change the relevant state
    someState = myEvent.Data;

    return myEvent;
}

重播时，您的事件溯源基础架构将调用 On(MyEvent) 以获得 MyEvent。既然你有一个 event 这意味着它是一个有效的状态转换 并且可以简单地应用；否则在您的初始 命令 处理中出现问题，您可能有错误。

事件存储中的所有事件都将按时间顺序进行聚合。除此之外，事件应该有一个全局序列号，以便于投影处理。

您可以有一个接受任何/所有事件的通用投影，然后将事件发布到服务总线上以进行系统集成。您还可以将这个负担放在事件存储的客户端上，让它自己跟踪位置，然后从存储本身读取事件。您可以结合这些并让客户端订阅服务总线事件，但通过跟踪位置（全局序列号）本身并在处理事件时更新它来确保它以相同的顺序执行它们。