MassTransit状态机Saga负载均衡下重复消费发布消息的解决方法

核心问题出在事件路由和Saga关联配置错误，导致Publish的事件被所有Saga实例接收并处理。以下是具体解决步骤：

1. 调整Saga状态机的事件关联配置 #

确保Saga通过唯一标识（如OrderId）关联所有相关事件，只有匹配该标识的Saga实例才会处理事件。同时在状态机中明确处理TaskCompleted和TaskFaulted事件：

// Saga状态实体，需包含关联ID和状态字段
public class MsrAutomationSagaState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
    public string OrderId { get; set; } // 与事件中的OrderId对应
    // 其他业务状态字段...
}

// 状态机定义
public class MsrAutomationStateMachine : MassTransitStateMachine<MsrAutomationSagaState>
{
    public MsrAutomationStateMachine(IWfTaskExecHandler wfTaskExecHandler, IWfManagementClient wfManagementClient)
    {
        InstanceState(x => x.CurrentState);

        // 初始事件：用OrderId作为关联ID
        Event(() => WfExecRequest, x =>
        {
            x.CorrelateById(context => context.Message.OrderId);
            x.OnMissingInstance(m => m.InitializeSagaState());
        });

        // TaskCompleted事件：关联同一个OrderId，无对应实例则忽略
        Event(() => TaskCompleted, x =>
        {
            x.CorrelateById(context => context.Message.OrderId);
            x.OnMissingInstance(m => m.Ignore());
        });

        // TaskFaulted事件：同理配置关联
        Event(() => TaskFaulted, x =>
        {
            x.CorrelateById(context => context.Message.OrderId);
            x.OnMissingInstance(m => m.Ignore());
        });

        // 状态转换逻辑
        Initially(
            When(WfExecRequest)
                .Then(context => { /* 初始化Saga状态 */ })
                .TransitionTo(Processing)
        );

        During(Processing,
            When(TaskCompleted)
                .Then(context => { /* 处理任务完成逻辑 */ })
                .TransitionTo(Completed),
            When(TaskFaulted)
                .Then(context => { /* 处理任务失败逻辑 */ })
                .TransitionTo(Faulted)
        );
    }

    // 定义状态和事件
    public State Processing { get; private set; }
    public State Completed { get; private set; }
    public State Faulted { get; private set; }

    public Event<IWfExecRequest> WfExecRequest { get; private set; }
    public Event<ITaskCompleted> TaskCompleted { get; private set; }
    public Event<ITaskFaulted> TaskFaulted { get; private set; }
}

2. 重构MassTransit配置，共享单个Saga接收端点 #

移除独立的WfTaskCompleted接收端点，将所有Saga相关事件路由到同一个队列，让多实例共享该队列实现负载均衡：

services.AddMassTransit(x =>
{
    // 添加Saga状态机和持久化存储（替换为你的实际存储，如EF/Redis）
    x.AddSagaStateMachine<MsrAutomationStateMachine, MsrAutomationSagaState>()
        .EntityFrameworkRepository(r =>
        {
            r.ConcurrencyMode = ConcurrencyMode.Pessimistic;
            r.AddDbContext<DbContext, MsrAutomationSagaDbContext>((provider, builder) =>
            {
                builder.UseSqlServer(provider.GetRequiredService<IConfiguration>().GetConnectionString("SagaDb"));
            });
        });

    // 保留原有Consumer（如果Saga未完全替代其逻辑）
    x.AddConsumer<WfExecRequestConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(HostCredets);

        // 单个接收端点处理所有Saga相关事件
        cfg.ReceiveEndpoint(queueName: "msr-automation-wf-saga", configureEndpoint: e =>
        {
            e.PrefetchCount = 1;
            // 配置Saga到该端点
            e.ConfigureSaga<MsrAutomationSagaState>(context);
            // 配置Consumer到同一端点（如果需要）
            e.ConfigureConsumer<WfExecRequestConsumer>(context);
        });

        // 移除原有的WfTaskCompleted接收端点
    });
});

3. 确保发布事件时携带正确的关联标识 #

发布TaskCompleted/TaskFaulted事件时，必须包含与初始请求一致的OrderId，让MassTransit能定位到对应的Saga实例：

// 发布TaskCompleted事件示例
await context.Publish<ITaskCompleted>(new
{
    OrderId = context.Message.OrderId, // 与初始请求的OrderId保持一致
    Timestamp = DateTime.UtcNow,
    // 其他业务字段...
}, context =>
{
    // 可选：显式设置CorrelationId，与OrderId映射
    context.CorrelationId = Guid.Parse(context.Message.OrderId);
});

解决原理 #

1. 共享队列负载均衡：多Saga实例监听同一个RabbitMQ队列，RabbitMQ会自动将消息分发到其中一个实例，避免所有实例接收同一消息。
2. CorrelationId关联：通过OrderId关联所有事件，MassTransit会从持久化存储中找到对应的Saga实例，只有该实例会处理事件，无匹配实例则忽略。
3. 统一事件路由：所有Saga相关事件都路由到同一端点，避免独立端点导致的消息广播问题。

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