在持久化和发布事件时处理最终的一致性

Question

目前正在开发一个DDD应用程序，该应用程序使用带有redis的事件源作为我的主要持久性存储。因此，不幸的是，如果发生故障，我没有内置回滚。应用程序是一个具有三个独立的骨料根的单体。

以下服务方法将

1. 创造我的“旅行”
2. 调用store来持久化与创建聚合相关的所有事件。
3. 调用快照回购以保持聚合快照
4. 发布事件以更新monolith中的其他聚合(保持松散耦合)

我使用的是c#，但是这里的语言与此无关：

public async Task<bool> CreateTripAsync(CreateTripRequest request)
{
    //Factory method to create Trip aggregate
    var trip = Trip.Create(request);

    //Persist trip events
    await _tripEventRepository.SaveEventsAsync(trip.Events);

    //Persist trip snapshot
    try
    {
        await _tripSnapshotRepository.SaveAsync(liveTrip);
    }
    catch
    {
        //Deal with eventual consistency by replaying stored events?? How do we trigger this to happen
    }

    //Publish all events associated with creating trip (e.g. 'PassengerAddedEvent') which will affect other aggregates in the same service
    // Mediators publish is fire and forget 
    var tasks = trip.Events.Select(async (ev) => { await _mediator.Publish(ev); });
    await Task.WhenAll(tasks);

    return true;
}

我的问题是：

1. 我的服务是做以上四件事。所有这些都与创建一次旅行相关。这会违反这种方法的SRP吗？我不这么认为，因为它们似乎都有相同的抽象级别。
2. 如果在事件存储中存储我的事件成功，但突然出现基础设施中断，而我无法更新快照，那么如何处理这种不一致？我正在考虑安排一个后台任务，该任务会重放与这次旅行相关的事件，以重新创建聚合，并且当它恢复在线时，将其保存。但是，假设我超过了我设定的最高重试数？
3. 因为我是在monolith中工作，所以我使用的是MediatR nuget包，它是一个火，忘记了pub子事件调度/处理程序。如果在发布这些事件处理程序时出现错误，我将使聚合处于不一致的状态。我知道，大多数企业级服务总线都有消息队列，这些消息队列会在异常后自动重试/存储。但是，我无法找到将ESB集成到monolith中的任何东西，在我的情况下，ESB可能会造成过度的杀伤力吗？

Answer 1

这会违反这种方法的SRP吗？

不值得担心。您可以合理地问，这段代码是否真的需要理解“保存”意味着“更新三个不同的存储”，或者这是否应该被抽象到另一个函数后面。例如，很多不同的Trip行为都需要一个保存--如果他们都有自己的保存协议副本，或者应该共享一个共同的协议。

您已经将域关注点与持久化关注点分离开来，这是最重要的。

如果在事件存储中存储我的事件成功，但突然之间，它们是基础设施中断，而我无法更新快照，我如何处理这种不一致。

更清楚地了解时间/时钟，以及如何使快照的使用者具有弹性。例如，如果快照使用描述快照在事件流中获取的位置的元数据保存，那么以后只使用新的事件就可以修复快照。

如果我正在查看的历史记录有100个事件，并且我的快照是根据事件95构建的，那么使用新事件更新快照是很容易的。所以，如果更新快照的尝试失败了，那就没什么大不了的。

这在本质上与数据库预写日志的工作方式相同。

如果在发布这些事件处理程序时出现错误，我将使聚合处于不一致的状态。

您可以稍后重新发布事件--特别是如果您设计您的聚合，以便它们能够识别他们已经看到的消息(又名:幂等事件处理)。也就是说，当聚合发生变化时，您可以同时安排两个发布事件，也可以按计划安排。这使您至少交付了一次，您只需获得处理副本的聚合即可。

另见没有人需要可靠的消息。