MassTransit中实现Singleton Saga并忽略同CorrelationId后续事件的问题

你的需求完全可以实现，当前报错是因为高并发下相同CorrelationId的事件同时到达时，多个请求同时尝试插入同_id的saga文档触发MongoDB唯一键约束，且你未配置对这类重复插入异常的自动处理逻辑，具体修复步骤如下：

问题根因 #

你当前的配置仅声明了无匹配saga时插入新实例，但没有覆盖两类异常场景：

1. 高并发下两个同CorrelationId的事件同时触发插入逻辑，第二个插入请求触发E11000重复键错误
2. 已存在对应saga实例时，没有配置逻辑忽略后续到达的同类型事件

修复步骤 #

1. 调整状态机事件处理逻辑 #

仅在saga初始化阶段处理GetOrdersIntegrationEventReceived事件，不要在其他状态分支声明对该事件的处理，这样已有saga实例收到该事件时会被直接忽略：

// 事件配置可以简化，SelectId可省略，CorrelateById默认会绑定saga的CorrelationId为_id
Event(() => GetOrdersIntegrationEventReceived, e =>
    e
        .CorrelateById(x => x.Message.CorrelationId)
        .SetSagaFactory(ctx => new GetOrdersState
        {
            CorrelationId = ctx.Message.CorrelationId,
            PartnerId = ctx.Message.PartnerId,
            LastUpdatedBeforeFromRequest = ctx.Message.LastUpdatedBefore
        })
        .InsertOnInitial = true);

// 状态处理分支：仅Initially阶段处理该事件
Initially(
    When(GetOrdersIntegrationEventReceived)
        // 你需要的其他初始化逻辑
        .TransitionTo(/* 你的自定义后续状态，例如OrdersSubmitted */)
);

// 不要在During、DuringAny等其他分支添加对GetOrdersIntegrationEventReceived的处理逻辑

2. 开启MongoDB Saga 仓库并发异常处理 #

MassTransit MongoDB Saga 集成已经内置了对E11000重复键异常的处理逻辑，只需要在注册Saga仓库时启用对应的配置，捕获到重复插入异常时会自动认为saga已创建，直接忽略当前事件，不会抛出异常：

services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<GetOrdersStateMachine, GetOrdersState>()
        .MongoDbRepository(r =>
        {
            r.ConnectionString = "你的MongoDB连接字符串";
            r.DatabaseName = "Dispatcher";
            r.CollectionName = "get.orders.states";
            // 关键配置：启用MongoDB并发异常处理
            r.UseMongoDbConcurrencyException();
        });
});

效果验证 #

调整配置后：

• 唯一CorrelationId的事件首次到达时会正常创建saga实例
• 后续同CorrelationId的事件到达时，如果saga已存在，因为没有匹配的处理分支会被直接忽略，不会修改原有saga状态
• 高并发场景下的重复插入错误会被自动捕获忽略，不会抛出异常

内容的提问来源于stack exchange，提问作者Marek M.