events - CQRS 中的命令处理程序、聚合、存储库和事件存储之间的关系

Question

我想了解基于 CQRS 的系统中命令处理程序、聚合、存储库和事件存储之间关系的一些细节。

到目前为止我所理解的:

命令处理程序从总线接收命令。他们负责从存储库加载适当的聚合并调用聚合上的域逻辑。完成后，他们从总线上删除命令。

聚合提供行为和内部状态。国家从不公开。改变状态的唯一方法是使用行为。建模此行为的方法从命令的属性创建事件，并将这些事件应用于聚合，然后调用相应地设置内部状态的事件处理程序。

存储库只允许在给定 ID 上加载聚合，并添加新聚合。基本上，存储库将域连接到事件存储。

最后但并非最不重要的一点是，事件存储负责将事件存储到数据库(或使用的任何存储)，并将这些事件重新加载为所谓的事件流。

到现在为止还挺好。
现在有一些我还没有得到的问题:

如果命令处理程序要在现有聚合上调用行为，那么一切都很容易。命令处理程序获取对存储库的引用，调用其 loadById 方法并返回聚合。但是，当还没有聚合但应该创建聚合时，命令处理程序会做什么？据我了解，以后应该使用这些事件重建聚合。这意味着聚合的创建是为了响应 fooCreated 事件而完成的。但是为了能够存储任何事件(包括 fooCreated 事件)，我需要一个聚合。所以这在我看来就像一个先有鸡还是先有蛋的问题:没有事件我无法创建聚合，但应该创建事件的唯一组件是聚合。所以基本上它归结为:我如何创建新的聚合，谁做什么？

当聚合触发事件时，内部事件处理程序会对其进行响应(通常通过应用方法调用)并更改聚合的状态。这个事件是如何移交给存储库的？谁发起了“请将新事件发送到存储库/事件存储”操作？聚合体本身？通过观察聚合存储库？订阅内部事件的其他人？ ...？

最后但并非最不重要的一点是，我在正确理解事件流的概念时遇到了问题:在我的想象中，它就像一个有序的事件列表。重要的是它是“有序的”。这是正确的吗？

Answer 1

以下内容基于我自己的经验以及我对 Lokad.CQRS、NCQRS 等各种框架的实验。我确信有多种方法可以处理这个问题。我会发布对我最有意义的内容。
1.聚合创建:
每次命令处理程序需要聚合时，它都会使用存储库。存储库从事件存储中检索相应的事件列表并调用重载的构造函数，注入(inject)事件

var stream = eventStore.LoadStream(id)
var User = new User(stream)

如果聚合之前不存在，则流将为空，并且新创建的对象将处于其原始状态。您可能希望确保在此状态下仅允许少数命令使聚合生效，例如User.Create() .
2.新事件的存储
命令处理发生在工作单元内。在命令执行期间，每个生成的事件都将添加到聚合内的列表中 (User.Changes)。执行完成后，更改将附加到事件存储中。在下面的示例中，这发生在以下行中:

store.AppendToStream(cmd.UserId, stream.Version, user.Changes)

3. 事件顺序
想象一下，如果两个后续 CustomerMoved事件以错误的顺序重播。
一个例子
我将尝试用一段伪代码来说明(我故意将存储库问题留在命令处理程序中以显示幕后会发生什么):
申请服务:

UserCommandHandler
    Handle(CreateUser cmd)
        stream = store.LoadStream(cmd.UserId)
        user = new User(stream.Events)
        user.Create(cmd.UserName, ...)
        store.AppendToStream(cmd.UserId, stream.Version, user.Changes)

    Handle(BlockUser cmd)
        stream = store.LoadStream(cmd.UserId)
        user = new User(stream.Events)
        user.Block(string reason)
        store.AppendToStream(cmd.UserId, stream.Version, user.Changes)

聚合:

User
    created = false
    blocked = false

    Changes = new List<Event>

    ctor(eventStream)
        isNewEvent = false
        foreach (event in eventStream)
            this.Apply(event, isNewEvent)

    Create(userName, ...)
        if (this.created) throw "User already exists"
        isNewEvent = true
        this.Apply(new UserCreated(...), isNewEvent)

    Block(reason)
        if (!this.created) throw "No such user"
        if (this.blocked) throw "User is already blocked"
        isNewEvent = true
        this.Apply(new UserBlocked(...), isNewEvent)

    Apply(userCreatedEvent, isNewEvent)
        this.created = true
        if (isNewEvent) this.Changes.Add(userCreatedEvent)

    Apply(userBlockedEvent, isNewEvent)
        this.blocked = true
        if (isNewEvent) this.Changes.Add(userBlockedEvent)

更新:
作为旁注:Yves 的回答让我想起了几年前 Udi Dahan 的一篇有趣的文章:

Don’t Create Aggregate Roots