事件源Saga实现

Question

我已经写了一个事件源聚合，现在实现了一个事件源佐贺。。。我注意到这两者是相似的，并创建了一个事件源对象作为基类，它们都是从中派生的。

我在这里看过一个演示http://blog.jonathanoliver.com/cqrs-sagas-with-event-sourcing-part-ii-of-ii/但是觉得可能有一个问题，因为在进程崩溃的情况下，命令可能会丢失，因为命令的发送在写事务之外？

public void Save(ISaga saga)
{
    var events = saga.GetUncommittedEvents();
    eventStore.Write(new UncommittedEventStream
    {
        Id = saga.Id,
        Type = saga.GetType(),
        Events = events,
        ExpectedVersion = saga.Version - events.Count
    });

    foreach (var message in saga.GetUndispatchedMessages())
        bus.Send(message); // can be done in different ways

    saga.ClearUncommittedEvents();
    saga.ClearUndispatchedMessages();
}

相反，我使用的是Greg Young的EventStore，当我保存EventSourcedObject（聚合或saga）时，顺序如下：

1. 存储库获取新的MutatingEvents列表。
2. 将它们写入流。
3. 当流被写入并提交到流时，EventStore会触发新事件。
4. 我们监听来自EventStore的事件并在EventHandler中处理它们。

我正在实现传奇的两个方面：

1. 接收事件，事件可能会转换状态，而状态又可能发出命令
2. 在将来的某个时候（通过外部计时器服务）我们可以被呼叫回来

问题

>

据我所知，事件处理程序不应该发出命令（如果命令失败会发生什么？）-但是我可以接受上述内容吗，因为Saga是通过此事件代理控制命令创建（对事件的反应）的实际事物，并且命令发送的任何失败都可以在外部处理（在处理Command dEmittedFromSaga并在命令失败时重新发送的外部EventHandler中）？

或者我忘了包装事件并将本机命令和事件存储在同一个流中（与基类消息混合在一起-佐贺将同时使用命令和事件，聚合只使用事件）？

网上有没有其他参考资料来实现事件来源的萨加斯？我可以理智地检查我的想法吗？

下面是一些背景代码。

Saga发出要运行的命令（包装在CommandEmittedFromSaga事件中）

下面的命令包装在事件中：

public class CommandEmittedFromSaga : Event
{
    public readonly Command Command;
    public readonly Identity SagaIdentity;
    public readonly Type SagaType;

    public CommandEmittedFromSaga(Identity sagaIdentity, Type sagaType, Command command)
    {
        Command = command;
        SagaType = sagaType;
        SagaIdentity = sagaIdentity;
    }
}

Saga在将来的某个时候请求回调（AlarmRequestedBySaga事件）

报警回调请求被包装在事件旁边，并将在请求的时间或之后对事件进行回击和事件处理：

public class AlarmRequestedBySaga : Event
{
    public readonly Event Event;
    public readonly DateTime FireOn;
    public readonly Identity Identity;
    public readonly Type SagaType;

    public AlarmRequestedBySaga(Identity identity, Type sagaType, Event @event, DateTime fireOn)
    {
        Identity = identity;
        SagaType = sagaType;
        Event = @event;
        FireOn = fireOn;
    }
}

或者，我可以将命令和事件存储在相同的基类型消息流中

public abstract class EventSourcedSaga
{
    protected EventSourcedSaga() { }

    protected EventSourcedSaga(Identity id, IEnumerable<Message> messages)
    {
        Identity = id;

        if (messages == null) throw new ArgumentNullException(nameof(messages));

        var count = 0;

        foreach (var message in messages)
        {
            var ev = message as Event;
            var command = message as Command;

            if(ev != null) Transition(ev);
            else if(command != null) _messages.Add(command);
            else throw new Exception($"Unsupported message type {message.GetType()}");

            count++;
        }

        if (count == 0)
            throw new ArgumentException("No messages provided");

        // All we need to know is the original number of events this
        // entity has had applied at time of construction.
        _unmutatedVersion = count;
        _constructing = false;
    }

    readonly IEventDispatchStrategy _dispatcher = new EventDispatchByReflectionStrategy("When");
    readonly List<Message> _messages = new List<Message>();
    readonly int _unmutatedVersion;
    private readonly bool _constructing = true;
    public readonly Identity Identity;

    public IList<Message> GetMessages()
    {
        return _messages.ToArray();
    }

    public void Transition(Event e)
    {
        _messages.Add(e);
        _dispatcher.Dispatch(this, e);
    }

    protected void SendCommand(Command c)
    {
        // Don't add a command whilst we are in the constructor. Message
        // state transition during construction must not generate new
        // commands, as those command will already be in the message list.
        if (_constructing) return;

        _messages.Add(c);
    }

    public int UnmutatedVersion() => _unmutatedVersion;
}

Answer 1

我相信前两个问题是对流程经理错误理解的结果（又名萨加斯，见底部术语注释）。

看起来你正试图将其建模为一个反向聚合（就像我曾经做的那样）。问题是：聚合的“社会契约”是其输入（命令）可以随时间变化（因为系统必须能够随时间变化），但其输出（事件）不能。一旦编写，事件就是历史问题，系统必须始终能够处理它们。有了这个条件，就可以从不可变的事件流可靠地加载聚合。

如果您尝试作为process manager实现反转输入和输出，则其输出不能是记录问题，因为随着时间的推移，命令可能会被弃用并从系统中删除。当您尝试用删除的命令加载流时，它将崩溃。因此，建模为反向聚合的流程管理器无法从不变的消息流可靠地重新加载。（嗯，我相信你可以想出一个办法……但这明智吗？）

因此，让我们通过查看流程管理器所替代的内容来考虑实施流程管理器。以管理订单履行等流程的员工为例。您为该用户做的第一件事是在UI中设置一个视图，供他们查看。您要做的第二件事是在UI中制作按钮，让用户根据他们在视图上看到的内容执行操作。例如：“此行已付款失败，因此我单击取消订单。此行已付款成功和订单项缺货，因此我单击更改订单。此订单已挂起，并且已过期1天，因此我单击标记订单进行审查。”。。。等等一旦决策过程定义明确并且开始需要用户太多的时间，您的任务就是将此过程自动化。为了实现自动化，其他一切都可以保持不变（视图，甚至一些用户界面，以便您可以检查），但用户已经变成了一段代码。

“走开，否则我会用一个非常小的shell脚本替换你。”

process manager代码现在定期读取视图，如果存在某些数据条件，则可能会发出命令。本质上，流程管理器的最简单版本是一些在计时器上运行的代码（例如，每小时运行一次），并且取决于特定的视图。那是我开始的地方。。。使用您已有的内容（视图/视图更新程序）和最少的添加（定期运行的代码）。即使您稍后决定需要针对某些用例使用不同的功能，“未来的您”将更好地了解需要解决的具体缺点。

这是一个很好的地方来提醒你高尔定律，也许还有雅格尼。

好的材料很难找到，因为这些概念有非常可塑性的实现，而且有各种各样的例子，其中许多是为了一般目的而过度设计的。然而，这里有一些我在回答中使用的参考资料。

DDD-不断发展的业务流程

请注意，术语Saga与流程管理器有不同的含义。一个常见的saga实现基本上是一个路由条，每个步骤及其相应的故障补偿都包含在该条上。这取决于路由条的每个接收者执行路由条上指定的内容并成功地将其传递到下一跳，或者执行故障补偿并向后路由。当处理由不同组管理的多个系统时，这可能有点过于乐观，因此经常使用流程管理器来代替。有关更多信息，请参阅此SO问题。