共享EventStore的CQRS+ES事件驱动系统事件重放问题咨询

你的核心问题在于事件处理器同时承担了投影更新和外部动作触发两个职责，而事件重放的唯一目的是重建投影视图，不需要重复执行发送邮件、通知这类有副作用的操作。以下是几种可行的解决方案：

方案1：拆分处理器职责，区分实时流与重放流 #

将原有的OrderCreatedEventHandler拆分为两个独立的处理器：一个专门负责更新投影（重放和实时流都执行），另一个仅负责触发外部动作（仅在实时事件流中执行，重放时跳过）。

投影更新处理器（重放/实时均执行） #

public class OrderCreatedProjectionHandler : INotificationHandler<OrderCreatedEvent>
{
    private readonly IOrderProjectionRepository _projectionRepo;

    public OrderCreatedProjectionHandler(IOrderProjectionRepository projectionRepo)
    {
        _projectionRepo = projectionRepo;
    }

    public async Task Handle(OrderCreatedEvent orderEvent, CancellationToken cancellationToken)
    {
        // 仅执行投影更新逻辑，无副作用
        await _projectionRepo.UpdateOrderProjection(orderEvent.Id, orderEvent.OrderDetails);
    }
}

外部动作处理器（仅实时流执行） #

通过事件上下文标记区分重放状态，重放时直接跳过：

public class OrderCreatedActionHandler : INotificationHandler<OrderCreatedEvent>
{
    private readonly IMediator _mediator;
    private readonly IEventExecutionContext _executionContext;

    public OrderCreatedActionHandler(IMediator mediator, IEventExecutionContext executionContext)
    {
        _mediator = mediator;
        _executionContext = executionContext;
    }

    public async Task Handle(OrderCreatedEvent orderEvent, CancellationToken cancellationToken)
    {
        // 检查是否处于重放模式，是则跳过
        if (_executionContext.IsReplayMode)
            return;

        await _mediator.Send(new SendShippingNotificationCommand(orderEvent.Id));
        await _mediator.Send(new SendOrderEmailCommand(orderEvent.Id));
    }
}

其中IEventExecutionContext是自定义的上下文服务，可在重放任务启动时设置IsReplayMode = true，重放结束后恢复为false；也可以从EventStore读取事件时，通过事件元数据（比如metadata["IsReplay"]）判断是否为重放事件。

方案2：为外部动作添加幂等性保障 #

通过EventStore记录外部动作的执行状态，每次执行前先检查是否已完成，避免重复操作。

public class OrderCreatedEventHandler : INotificationHandler<OrderCreatedEvent>
{
    private readonly IMediator _mediator;
    private readonly IEventStore _eventStore;

    public OrderCreatedEventHandler(IMediator mediator, IEventStore eventStore)
    {
        _mediator = mediator;
        _eventStore = eventStore;
    }

    public async Task Handle(OrderCreatedEvent orderEvent, CancellationToken cancellationToken)
    {
        // 检查是否已发送过发货通知
        var hasSentShippingNotify = await _eventStore.ExistsAsync<ShippingNotificationSentEvent>(orderEvent.Id);
        if (!hasSentShippingNotify)
        {
            await _mediator.Send(new SendShippingNotificationCommand(orderEvent.Id));
            // 记录动作完成事件到EventStore
            await _eventStore.AppendEventAsync(new ShippingNotificationSentEvent(orderEvent.Id));
        }

        // 检查是否已发送过订单邮件
        var hasSentOrderEmail = await _eventStore.ExistsAsync<OrderEmailSentEvent>(orderEvent.Id);
        if (!hasSentOrderEmail)
        {
            await _mediator.Send(new SendOrderEmailCommand(orderEvent.Id));
            await _eventStore.AppendEventAsync(new OrderEmailSentEvent(orderEvent.Id));
        }
    }
}

这种方式不仅解决重放问题，还能处理消息重试、重复投递等场景下的幂等性问题，是生产环境中推荐的做法。

方案3：使用专用投影重建管道 #

完全绕过常规的EventBus订阅处理器，编写专门的投影重建服务，直接从EventStore读取事件并更新投影，不触发任何业务命令。

public class OrderProjectionRebuilder
{
    private readonly IEventStore _eventStore;
    private readonly IOrderProjectionRepository _projectionRepo;

    public OrderProjectionRebuilder(IEventStore eventStore, IOrderProjectionRepository projectionRepo)
    {
        _eventStore = eventStore;
        _projectionRepo = projectionRepo;
    }

    public async Task RebuildAsync(CancellationToken cancellationToken)
    {
        // 清空现有投影（可选，根据需求）
        await _projectionRepo.ClearAllAsync(cancellationToken);

        // 读取所有OrderCreatedEvent
        var orderEvents = await _eventStore.ReadAllEventsAsync<OrderCreatedEvent>(cancellationToken);

        foreach (var evt in orderEvents)
        {
            // 仅执行投影更新
            await _projectionRepo.UpdateOrderProjection(evt.Id, evt.OrderDetails);
        }
    }
}

这种方式隔离了重放逻辑与正常业务流程，完全避免了副作用触发的风险，适合大规模投影重建场景。

内容的提问来源于stack exchange，提问作者deviantxdes