.NET Core中MassTransit Saga状态机结合Azure Service Bus Send消息问题

当前实现的问题 #

1. 队列冗余原因：使用Publish时，MassTransit会为每个事件类型创建独立队列（每个消费者对应一个队列），这就是你生成11个队列的核心原因。
2. Send方法失效核心：直接Send到Topic订阅时，未确保事件能被Saga状态机接收——状态机需要监听对应订阅，且事件的CorrelationId必须正确传递，否则状态机无法匹配到对应的实例。
3. 流程逻辑偏差：你期望OrderRequestedConsumer执行完成后触发OrderInitiatedEvent，同时驱动状态机流转，但当前代码中状态机是收到OrderRequestedEvent后直接PublishStockRequested，未与OrderInitiatedConsumer的执行结果关联。

基于Azure Service Bus Topic-Subscription的实现方案 #

1. 全局Topic规划 #

创建一个共享Topic（例如OrderProcessingTopic），所有业务事件都发送到该Topic，为不同消费者（含Saga状态机）创建独立订阅，通过SQL过滤器控制每个订阅仅接收自身关心的事件类型。

2. 业务消费者（Order/Stock/Billing）的订阅配置 #

以Order服务为例，注册消费者时不再创建独立队列，而是绑定到共享Topic的订阅并添加过滤器：

// Order服务配置代码
cfg.SubscribeToTopic("OrderProcessingTopic", nameof(OrderRequestedConsumer), x =>
{
    // SQL过滤器：仅接收OrderRequestedEvent类型消息
    x.Filter = new SqlFilter($"MessageType = '{typeof(OrderRequestedEvent).FullName}'");
    x.ConfigureConsumer<OrderRequestedConsumer>(context);
});

cfg.SubscribeToTopic("OrderProcessingTopic", nameof(OrderInitiatedConsumer), x =>
{
    x.Filter = new SqlFilter($"MessageType = '{typeof(OrderInitiatedEvent).FullName}'");
    x.ConfigureConsumer<OrderInitiatedConsumer>(context);
});

Stock、Billing服务的消费者配置逻辑一致，每个消费者对应Topic的一个订阅，通过过滤器筛选目标事件。

3. Saga状态机的订阅配置 #

状态机需要监听所有关联事件（如OrderInitiatedEvent、StockCompletedEvent等），为此创建一个专属订阅，用多条件过滤器包含所有需处理的事件类型：

// Saga状态机配置代码
cfg.SubscribeToTopic("OrderProcessingTopic", "OrderSagaSubscription", x =>
{
    // 多条件过滤器：包含状态机需处理的所有事件类型
    x.Filter = new SqlFilter($"MessageType = '{typeof(OrderRequestedEvent).FullName}' OR " +
                            $"MessageType = '{typeof(OrderInitiatedEvent).FullName}' OR " +
                            $"MessageType = '{typeof(StockCompletedEvent).FullName}' OR " +
                            $"MessageType = '{typeof(OrderFailedEvent).FullName}'");
    x.ConfigureSaga<OrderSagaInstance>(context);
});

4. Send方法的正确使用方式 #

发送事件时直接Send到共享Topic，确保MessageType属性正确（MassTransit自动处理），同时传递CorrelationId关联Saga实例：

（1）OrderRequestedConsumer中发送OrderInitiatedEvent

public class OrderRequestedConsumer : IConsumer<OrderRequestedEvent>
{
    private readonly ISendEndpointProvider _sendEndpointProvider;

    public OrderRequestedConsumer(ISendEndpointProvider sendEndpointProvider)
    {
        _sendEndpointProvider = sendEndpointProvider;
    }

    public async Task Consume(ConsumeContext<OrderRequestedEvent> context)
    {
        // 执行OrderRequested业务逻辑
        Console.WriteLine($"处理订单请求: {context.Message.OrderId}");

        // 发送OrderInitiatedEvent到共享Topic
        var topicEndpoint = await _sendEndpointProvider.GetSendEndpoint(new Uri($"topic:OrderProcessingTopic"));
        await topicEndpoint.Send(new OrderInitiatedEvent
        {
            CorrelationId = context.Message.CorrelationId,
            OrderId = context.Message.OrderId
        });
    }
}

（2）Saga状态机处理OrderInitiatedEvent并发送StockRequestedEvent

修正状态机逻辑，确保收到OrderInitiatedEvent后流转状态并发送StockRequestedEvent：

During(OrderStates.OrderRequested,
    When(OrderEvents.OrderRequestedEvent)
        .Then(context => Console.WriteLine($"订单已请求: {context.Instance.OrderId}")),
    When(OrderEvents.OrderInitiatedEvent)
        .Then(context => Console.WriteLine($"订单已初始化: {context.Instance.OrderId}"))
        .SendAsync(async context =>
        {
            var topicEndpoint = await context.GetSendEndpoint(new Uri($"topic:OrderProcessingTopic"));
            return new SendRequest<StockRequestedEvent>(topicEndpoint, new StockRequestedEvent
            {
                CorrelationId = context.Instance.CorrelationId,
                OrderId = context.Instance.OrderId
            });
        })
        .TransitionTo(OrderStates.OrderInitiated),
    When(OrderEvents.OrderFailedEvent)
        .Then(context => Console.WriteLine($"订单处理失败: {context.Instance.OrderId}"))
        .TransitionTo(OrderStates.OrderFailed)
);

5. 关键注意事项 #

• CorrelationId一致性：所有事件必须携带相同的CorrelationId，确保Saga状态机能匹配到对应实例。
• MessageType正确性：Azure Service Bus过滤器依赖MessageType属性，MassTransit默认会写入消息类型全名，请勿手动修改。
• AKS部署适配：在AKS中部署时，通过Secret管理Service Bus连接字符串，用环境变量注入MassTransit配置；每个微服务Pod仅监听对应Topic订阅，无需创建大量队列，降低运维复杂度。
• 错误处理：为每个订阅配置死信队列（DLQ），处理消费失败的消息，避免影响整体流程。

队列优化效果 #

采用该方案后，所有事件通过单个共享Topic传递，每个消费者对应一个订阅、Saga对应一个订阅，总订阅数等于消费者数+Saga数，相比原方案的事件队列模式，数量大幅减少且便于统一管理。

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