cqrs - 如何从数据库“EventStore”中持久保存聚合/读取模型？

Question

第一次尝试实现事件源和CQRS，但在持久化聚合时陷入困境。

这就是我现在的位置


我已经将“ EventStore”设置为流，“ foos”
从node-eventstore-client连接到它
我订阅追赶活动


一切都很好。

借助eventAppeared事件处理程序功能，无论何时发生事件，我都可以构建聚合。很好，但是我该怎么办？

假设我建立并汇总了Foos列表

    [
      {
        id: 'some aggregate uuidv5 made from barId and bazId',
        barId: 'qwe',
        bazId: 'rty',
        isActive: true,
        history: [
          {
            id: 'some event uuid',
            data: {
              isActive: true,
            },
            timestamp: 123456788,
            eventType: 'IsActiveUpdated'
          }
          {
            id: 'some event uuid',
            data: {
              barId: 'qwe',
              bazId: 'rty',
            },
            timestamp: 123456789,
            eventType: 'FooCreated'
          }
        ]
      }
    ]

为了遵循CQRS，我将在读取模型中构建以上汇总，对吗？但是，如何将该聚合存储在数据库中？

我想只是一个nosql数据库就可以了，但是我肯定需要一个数据库，因为我会将gRPC APi放在此数据库和其他读取模型/聚集的前面。

但是，从构建聚合到将其持久保存到数据库中，我实际上是怎么做的呢？

我曾经尝试按照本教程 https://blog.insiderattack.net/implementing-event-sourcing-and-cqrs-pattern-with-mongodb-66991e7b72be进行操作，该教程非常简单，因为您将mongodb都用作事件存储，并且只创建了一个聚合视图并在收到新事件时更新该视图。它有缺陷和局限性（聚合管道），这就是为什么我现在将事件存储部分转向“ EventStore”。

但是，如何持久保存当前仅由“ EventStore”中的事件构建并存储在代码/内存中的聚合……？

我觉得这可能是一个愚蠢的问题，但是我是否必须遍历数组中的每个项目并将每个项目插入db表/集合中，或者您是否有某种方式可以一次将整个数组/聚合转储在那里？

之后会怎样？您是否为每个集合创建一个物化视图并对此进行查询？

我愿意为此选择最好的数据库，无论是postgres /其他rdbms，mongodb，cassandra，redis，表存储等。

最后一个问题。现在，我仅使用单个流“ foos”，但在此级别上，我希望新事件会频繁发生（每隔几秒钟左右），但据我了解，您仍然会坚持并使用实例化对其进行更新意见对吗？

因此，鉴于可以结合使用barId和bazId来对事件进行分组，而不是单个流，我认为更专业的流（例如foos-barId-bazId）将是您可以尝试的方法，以减少传入新消息的频率事件到重新创建实体化视图的意义。

是否有一般的经验法则说，如果更新频率低于特定限制，则不重新创建/更新/刷新实例化视图？那么唯一的替代方法是从普通表/集合中查询吗？

编辑：

最后，我试图制作一个只有2个rpcs的gRPC api-一个用于通过id获取单个foo，一个用于获取所有foos（具有用于按状态过滤的可选字段-但这并不重要）。简化的原型看起来像这样：

rpc GetFoo(FooRequest) returns (Foo)
rpc GetFoos(FoosRequest) returns (FooResponse)

message FooRequest {
    string id = 1; // uuid
}

// If the optional status field is not specified, return all foos
message FoosRequest {
    // If this field is specified only return the Foos that has isActive true or false
    FooStatus status = 1;

    enum FooStatus {
        UNKNOWN = 0;
        ACTIVE = 1;
        INACTIVE = 2;
    }
}

message FoosResponse {
    repeated Foo foos;
}

message Foo {
    string id = 1; // uuid
    string bar_id = 2 // uuid
    string baz_id = 3 // uuid
    boolean is_active = 4;
    repeated Event history = 5;
    google.protobuf.Timestamp last_updated = 6;
}

message Event {
    string id = 1; // uuid
    google.protobuf.Any data = 2;
    google.protobuf.Timestamp timestamp = 3;
    string eventType = 4;
}

传入事件看起来像这样：

{
  id: 'some event uuid',
  barId: 'qwe',
  bazId: 'rty',
  timestamp: 123456789,
  eventType: 'FooCreated'
}

{
  id: 'some event uuid',
  isActive: true,
  timestamp: 123456788,
  eventType: 'IsActiveUpdated'
}

如您所见，在gRPC API中没有uuid可以使GetFoo（uuid）成为可能，这就是为什么我将生成带有barId和bazId的uuidv5的原因，将其组合在一起将是有效的uuid。我在上面看到的投影/汇总中进行了说明。

此外，GetFoos rpc要么返回所有foo（如果状态字段未定义），要么返回具有与状态字段匹配的isActive的foo（如果指定）。

但是我不知道如何从追赶订阅处理程序继续。

我将事件存储在“ EventStore”（ https://eventstore.com/）中，使用具有追赶功能的订阅，我以所需的形式构建了带有Foo数组的聚合/投影，但能够获得单个Foo根据我的gRPC API的ID，我想我需要将整个聚合/投影存储在某种数据库中，以便我可以连接并从gRPC API提取数据？每次出现新事件时，我都需要将该事件也添加到数据库中，或者它如何工作？

我想我已经阅读了我在互联网上可以找到的所有资源，但是仍然缺少一些关键信息来弄清楚这一点。

gRPC不是那么重要。我猜可能是REST，但是我最大的问题是如何使汇总/投影的数据可用于API服务（可能还需要更多API）？我想我将需要将具有生成的uuid和历史字段的聚合/投影数据存储在数据库中，以便能够通过uuid从API服务中获取数据，但是从catchup事件来看，是什么数据库以及该存储过程是如何完成的我在哪里建立集合的处理程序？

Answer 1

我知道你的感受！基本上这就是我第一次尝试进行CQRS和ES时发生的事情。

我认为您在知识上有一些差距，我相信您会很快弥补。您可以在执行操作时从事件流中添加聚合。那就是您的总和持续存在。读取模型有所不同。让我解释...

您的读取模型是您用来对之运行查询并提供数据以显示给UI的东西。您的聚合不（直接）参与其中。实际上，它们应该被封装。这意味着您无法从外部“看到”它们的状态。即没有getter和setter，但具有ID的聚合ID除外。

本文为您提供了有关如何将它们组合在一起的有用概述：CQRS + Event Sourcing – Step by Step

这个想法是，当一个聚合改变状态时，它只能通过它产生的一个事件来改变。您将该事件存储在事件存储中。该事件也会发布，以便可以更新读取的模型。

同时查看您的聚合，它看起来更像是典型的读取模型对象或DTO。聚合对功能而不是属性感兴趣。因此，您会期望看到无效的公共功能，无法向集合发出命令。但不是像isActive或history这样的公共属性。

我希望这是有道理的。

编辑：

这里有一些更实用的建议。

“为了遵循CQRS，我将在读取模型中构建以上汇总，对吗？”
您没有在读取模型中构建聚合。它们是等式的CQRS侧的不同侧的独立事物。聚合位于命令端。针对与聚合不同的读取模型进行查询。

集合具有公共无效函数，没有getter或setter（集合ID除外）。它们被封装。当状态由于发出命令而改变时，它们将生成事件。这些事件存储在事件存储中，并用于恢复聚合的状态。换句话说，这就是聚合的存储方式。

事件继续被发布，因此事件处理程序和其他进程可以对它们做出反应并更新读取模型和/或触发新的级联命令。

“最后一个问题。目前，我仅使用单个流“ foos”，但是在这个级别上，我希望新事件会非常频繁地发生（每隔几秒钟左右），但是据我了解，您仍然会坚持下去，并且使用实例化视图更新它，对吗？”

每两秒钟很可能就好了。我更关注使用物化视图的持久性和更新。我不知道您的意思是什么，但这听起来不像您有正确的主意。视图应该是非常简单的读取模型。无需像在RDMS中发现的那样复杂的关系。因此，经过高度优化，可以快速读取。