使用Azure Service Bus时遭遇MassTransit.RequestTimeoutException问题

作为MassTransit新手，我计划在Azure Service Bus中使用Durable Future与Routing Slip模式。Web应用已编写对应控制器代码，预期请求提交后由Activities类完成验证逻辑：验证不通过则向客户端返回错误，通过则继续后续流程。但切换到Azure Service Bus后，始终触发MassTransit的RequestTimeoutException（请求超时等待响应），改用RabbitMQ则能正常执行验证逻辑。想请教是否需要为请求客户端单独创建消费者，或是遗漏了关键配置步骤？

控制器代码 #

app.MapPost("/orders/submit", async (SubmitOrderDto submitOrderDto, IRequestClient<SubmitOrder> client, ILogger<Program> logger, CancellationToken cancellationToken) =>
{
    try
    {
        Response response = await client.GetResponse<OrderCompleted, OrderFaulted>(new
        {
            submitOrderDto.OrderId
        }, cancellationToken);
        return response switch
        {
            (_, OrderCompleted completed) => Results.Ok(new
            {
                completed.OrderId,
                completed.Status,
                completed.Created,
                completed.Completed,
            }),
            (_, OrderFaulted faulted) => Results.BadRequest(new
            {
                faulted.OrderId,
                faulted.Description,
                faulted.Created,
                faulted.Faulted
            }),
            _ => Results.BadRequest()
        };
    }
    catch (Exception ex)
    {
        return Results.Accepted(value:new
        {
            submitOrderDto.OrderId, 
            ex.Message
        });
    }
})
.Produces<OrderCompleted>()
.Produces<OrderFaulted>(StatusCodes.Status400BadRequest)
.Produces(StatusCodes.Status202Accepted)
.ProducesValidationProblem(StatusCodes.Status409Conflict)
.WithName("SubmitOrder").WithTags("OrderServiceAPI");

异常信息 #

Timeout waiting for response, RequestId: 40040000-4446-107b-e828-08db2e658cb0
at MassTransit.Clients.RequestClient1.GetResponseInternal[T1,T2](SendRequestCallback request, CancellationToken cancellationToken, RequestTimeout timeout, RequestPipeConfiguratorCallback1 callback) in /_/src/MassTransit/Clients/RequestClient.cs:line 212

Program配置代码 #

var builder = WebApplication.CreateBuilder(args);
builder.Services.TryAddScoped<IItineraryPlanner<SubmitOrder>, OrderItineraryPlannerAzure>();

builder.Services.AddMassTransit(x =>
{
    // Azure
    x.SetKebabCaseEndpointNameFormatter();
    x.AddServiceBusMessageScheduler();
    x.AddRequestClient<SubmitOrder>();

    //-------------
    x.ApplyCustomMassTransitConfiguration();
    x.AddDelayedMessageScheduler();
    x.AddActivitiesFromNamespaceContaining<OrderActivity>();
    x.AddFuturesFromNamespaceContaining<OrderFuture>();

    x.AddSagaRepository<FutureState>()
        .InMemoryRepository();

    //x.UsingRabbitMq((context, cfg) =>
    //{
    //    cfg.AutoStart = true;
    //    cfg.ApplyCustomBusConfiguration();

    //    cfg.UseDelayedMessageScheduler();
    //    cfg.ConfigureEndpoints(context);
    //});

    x.UsingAzureServiceBus((context, cfg) =>
    {
        using var scope = context.CreateScope();  
        cfg.Host("Endpoint=sb://{connectionString}");

        cfg.UseTimeout(timeoutConfigurator =>
        {
            timeoutConfigurator.Timeout = TimeSpan.FromMinutes(10);
        });
        cfg.PrefetchCount = 100;
        cfg.LockDuration = TimeSpan.FromMinutes(5);
        cfg.MaxConcurrentCalls = 100;
        cfg.MaxDeliveryCount = 5;
        cfg.DefaultMessageTimeToLive = TimeSpan.FromDays(7);
        cfg.AutoStart = true;
        cfg.RequiresSession = true;

        cfg.UseServiceBusMessageScheduler();
        // deploy topology configuration when service starting
        cfg.DeployTopologyOnly = true;
        cfg.UseDelayedMessageScheduler();
        cfg.ConfigureEndpoints(context);
    });
});

问题排查与解决方案 #

核心问题点 #

1. DeployTopologyOnly = true：这个配置仅创建Azure Service Bus的队列、主题等拓扑资源，但不会启动消息消费者。你的Activities和Durable Future处理逻辑完全没运行，请求发送后无人处理，最终超时。
2. RequiresSession = true：全局开启会话要求，但Durable Future和Routing Slip的默认端点不依赖会话，强制开启会导致消息无法被正确路由——RequestClient发送的请求不带对应会话ID，消息会卡在队列里无法处理。
3. 消息调度器重复配置：同时添加了Service Bus和内存延迟调度器，Azure Service Bus原生支持延迟消息，冗余配置可能引发冲突。

修复步骤 #

1. 删除DeployTopologyOnly = true：确保服务启动时自动启动消费者，处理消息请求。
2. 删除全局RequiresSession = true：仅在需要会话的特定端点单独配置，业务不需要的话直接移除。
3. 调整消息调度器配置：保留AddServiceBusMessageScheduler和cfg.UseServiceBusMessageScheduler()，删除AddDelayedMessageScheduler()和cfg.UseDelayedMessageScheduler()。
4. 显式配置RequestClient超时（可选）：给RequestClient设置足够长的超时，确保Activity处理有充足时间：

   x.AddRequestClient<SubmitOrder>(new RequestTimeout(TimeSpan.FromMinutes(15)));
   

5. 验证端点注册：确认ConfigureEndpoints(context)已正确注册所有Activity和Future的消费者端点，未被自定义配置覆盖。

修复后的Azure Service Bus配置示例 #

x.UsingAzureServiceBus((context, cfg) =>
{
    cfg.Host("Endpoint=sb://{connectionString}");

    cfg.UseTimeout(timeoutConfigurator =>
    {
        timeoutConfigurator.Timeout = TimeSpan.FromMinutes(10);
    });
    cfg.PrefetchCount = 100;
    cfg.LockDuration = TimeSpan.FromMinutes(5);
    cfg.MaxConcurrentCalls = 100;
    cfg.MaxDeliveryCount = 5;
    cfg.DefaultMessageTimeToLive = TimeSpan.FromDays(7);
    cfg.AutoStart = true;

    cfg.UseServiceBusMessageScheduler();
    cfg.ConfigureEndpoints(context);
});

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