events - CQRS 存储库/事件发布者

Question

我正在使用 CqrsLite对于 CQRS 风格的项目。具体 Repository 实现的 Save 方法看起来像这样(省略了不相关的行)。

    public void Save<T>(T aggregate, int? expectedVersion = null) where T : AggregateRoot
    {
        if (expectedVersion != null && _eventStore.Get(typeof(T), aggregate.Id, expectedVersion.Value).Any())
            throw new ConcurrencyException(aggregate.Id);

        var i = 0;
        foreach (var @event in aggregate.GetUncommittedChanges())
        {
            // ... [irrelevant code removed] ...
            _eventStore.Save(typeof(T), @event);
            _publisher.Publish(@event);
        }
        aggregate.MarkChangesAsCommitted();
    }

令我不安的是，此方法正在提交要发布给订阅者的事件，然后才告诉聚合将它们标记为已提交。因此，如果观察给定事件的事件处理程序阻塞，则聚合将不会提交可能已通知先前事件处理程序的更改。

为什么我不将 _publisher.Publish(@event) 移动到 aggregate.MarkChangesAsCommitted() 之后，就像这样。我错过了什么？

    public void Save<T>(T aggregate, int? expectedVersion = null) where T : AggregateRoot
    {
        if (expectedVersion != null && _eventStore.Get(typeof(T), aggregate.Id, expectedVersion.Value).Any())
            throw new ConcurrencyException(aggregate.Id);

        var events = aggregate.GetUncommittedChanges();
        foreach (var @event in events)
        {
            // ... [irrelevant code removed] ...
            _eventStore.Save(typeof(T), @event);
        }
        aggregate.MarkChangesAsCommitted();
        _publisher.Publish(events);
    }

Answer 1

这两种方法都有问题，因为 Save 之间可能存在错误。和 Publish ，无论这两个方法的调用顺序如何。这可能会导致发布未保存的事件或未发布已保存的事件。内存中状态损坏(在聚合对象中)的问题也存在(尽管可以通过简单地捕获事件处理程序产生的错误来处理)。

此问题的一种解决方案是使用两阶段提交(例如，如果您的事件存储基于 SQL Server 并且发布者基于 MSMQ，则可用)。但是，这对性能、可伸缩性和操作有影响，并且不允许延迟订阅者(见下文)。

更好的方法是允许对事件感兴趣的各方将它们从事件存储中提取出来(理想情况下，将此与某种通知机制或长轮询相结合以使其更具“ react 性”)。这将跟踪最后接收事件的责任转移到订阅者，允许

迟到的订阅者(在事件存储后很久才加入)以接收旧事件和新事件，

没有两阶段提交的可靠性。

在搜索诸如“将事件存储用作队列”之类的内容时，您应该找到有关此方法的更多信息，并且 Greg 的回答中的视频也可能对此有所帮助。

一种常见的算法是这样的:

事件存储为每个保存的事件分配一个检查点 token (例如，序列号)；

订阅者从他们知道的最后一个检查点 token (如果有)开始，向事件存储询问新事件(定期，基于长轮询，对推送通知使用react等)，

事件存储发送从该检查点 token 开始的更新事件以及新的检查点 token ，

订阅者处理事件，并在可能的情况下以原子方式存储新的检查点 token 以及它们产生的任何副作用；

如果无法进行原子保存，他们可以在产生副作用后存储新的检查点 token ，并且他们需要一种方法来忽略他们已经看到的事件，以防中间出现错误(事件处理被称为“幂等");

订阅者再次从#2 开始。

我想补充一点，我不考虑忽略 Save 的事件存储。/ Publish问题生产就绪。有关替代方案，请参阅 Greg Young's Event Store或(目前或多或少未维护) NEventStore .