如何结合Mass Transit请求响应与Azure Service Bus会话实现产品类型串行处理？

问题描述 #

我需要用Mass Transit和Azure Service Bus实现「添加产品」命令的请求响应模式，既要能发送请求并接收响应，还要让命令消费者根据「产品类型」属性处理请求——同一产品类型的请求不能并发处理。

举个例子：队列里有3种产品类型的多个请求，单个消费者的处理顺序应该是：

• 先处理完所有类型1的产品
• 再处理所有类型2的产品
• 最后处理所有类型3的产品

（产品类型的处理先后顺序无所谓）

我知道Azure Service Bus的会话功能可以实现这个串行需求，但不确定怎么将会话和Mass Transit的请求响应模式（也就是IRequestClient.GetResponse()和ConsumeContext.RespondAsync()配对使用）结合起来，也不清楚生产者和消费者两端该怎么配置。

我试过给Send操作设置会话，但看起来没和请求响应关联上，现有代码如下：

生产者端现有代码 #

x.UsingAzureServiceBus((context, cfg) =>
{
    cfg.Host(connectionString);
    cfg.Send<AddProductRequest>(c => c.UseSessionIdFormatter(s => s.Message.ProductTypeId.ToString("D")));
}); 

消费者端现有代码 #

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<SubmittedProductConsumer>();
    
    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host("myconnectionstring");

        cfg.ReceiveEndpoint("submit-product-worker", e =>
        {
            e.RequiresSession = true;
            
            e.ConfigureConsumer<SubmittedProductConsumer>(context);
        });
    });
});

解决方案 #

要实现带会话的请求响应模式，核心是让请求消息绑定到对应产品类型的会话，同时让响应能正确回传。之前的配置问题在于：请求响应模式下默认会使用临时响应队列，但你需要显式配置会话ID的传递，并且确保消费者的接收端点正确关联会话。

1. 生产者端正确配置 #

需要给AddProductRequest的请求客户端指定会话ID，同时明确目标队列（和消费者端的接收端点名称一致），确保会话ID随请求一起传递：

builder.Services.AddMassTransit(x =>
{
    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(connectionString);

        // 配置请求客户端，指定目标队列（与消费者端接收端点名称匹配）
        cfg.ConfigureRequestClient<AddProductRequest>(context, cfg =>
        {
            cfg.TargetAddress = new Uri($"queue:submit-product-worker");
            // 基于ProductTypeId生成会话ID，确保同类型请求进入同一会话
            cfg.UseSessionIdFormatter(req => req.Message.ProductTypeId.ToString("D"));
        });
    });
});

2. 消费者端正确配置 #

消费者的接收端点已设置RequiresSession = true，这部分是正确的，只需补充会话超时配置（避免会话长时间占用资源）即可：

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<SubmittedProductConsumer>();
    
    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host("myconnectionstring");

        cfg.ReceiveEndpoint("submit-product-worker", e =>
        {
            e.RequiresSession = true;
            e.ConfigureConsumer<SubmittedProductConsumer>(context);
            
            // 设置会话空闲超时，无新消息时自动释放会话
            e.SessionIdleTimeout = TimeSpan.FromMinutes(5);
        });
    });
});

3. 消费者业务逻辑实现 #

消费者只需正常处理请求并响应，Mass Transit会自动维护会话的串行处理逻辑：

public class SubmittedProductConsumer : IConsumer<AddProductRequest>
{
    public async Task Consume(ConsumeContext<AddProductRequest> context)
    {
        // 根据ProductTypeId执行添加产品的业务逻辑
        var productTypeId = context.Message.ProductTypeId;
        // ... 业务代码 ...
        
        // 响应请求
        await context.RespondAsync(new AddProductResponse
        {
            Success = true,
            ProductId = Guid.NewGuid()
        });
    }
}

关键说明 #

• 会话ID基于ProductTypeId生成，同一类型的请求会进入同一个会话，Azure Service Bus会保证同一会话内的消息串行处理，不同会话的消息可并发（若有多个消费者实例）。
• 请求响应模式下必须指定TargetAddress，否则Mass Transit会使用临时队列，无法关联到启用会话的固定队列。
• SessionIdleTimeout需合理设置，避免会话因长时间无消息占用资源。

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