MassTransit Saga从Azure Service Bus拉取消息为何存在1分钟延迟？

使用技术栈：MassTransit 8.3.4、.NET 7、Azure Service Bus「标准」定价层。

操作步骤 #

• 未运行saga时，向saga队列发送50条消息，每条消息携带唯一CorrelationId。
• 运行包含saga的ASP.NET Web API，采用默认并发设置（MaxConcurrentSessions=8），存储库使用启用会话的ASB。

现象 #

• 并行处理8条消息
• <1分钟停顿，无任何saga实例处理任务>
• 并行处理8条消息
• <1分钟停顿，无任何saga实例处理任务>
• 重复上述循环

疑问 #

为什么会出现1分钟延迟？遗漏了什么配置？saga实例处理完成后为何不立即从队列拉取新消息？

源代码 #

var builder = WebApplication.CreateBuilder(args);
builder.Services.AddControllers();

var queueAName = "commandA";
builder.Services.AddMassTransit(x =>
{
    x.AddSagaStateMachine<SimpleSaga, SimpleSagaState>().MessageSessionRepository();

    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.ReceiveEndpoint(queueAName, c =>
        {
            c.RequiresSession = true;
            c.ConfigureSaga<SimpleSagaState>(context);
        });

        cfg.Host("");
    });
});
var app = builder.Build();
app.UseHttpsRedirection();
app.UseAuthorization();
app.MapControllers();
app.Run();


public class SimpleSaga : MassTransitStateMachine<SimpleSagaState>
{
    public SimpleSaga() 
    {
        InstanceState(x => x.CurrentState);

        Event(() => InitialCommandReceived, x => x.CorrelateById(context => context.CorrelationId.Value));

        Initially(
            When(InitialCommandReceived)
                .Then(c => Console.WriteLine($"[{DateTime.UtcNow.TimeOfDay}]: InitialCommandReceived: received with CorrelationId: {c.CorrelationId}."))
                .Finalize());

        SetCompletedWhenFinalized();
    }

    public Event<CommandA> InitialCommandReceived { get; set; }
}

public class SimpleSagaState : SagaStateMachineInstance
{
    public Guid CorrelationId { get; set; }
    public string CurrentState { get; set; }
}

输出日志 #

[19:19:02.4539790]: InitialCommandReceived: received with CorrelationId: 538f04fd-62b1-4885-8be2-4f1e1a143539.
[19:19:02.4540137]: InitialCommandReceived: received with CorrelationId: 87077e87-447d-422e-b078-b73236c9d23e.
[19:19:02.4540267]: InitialCommandReceived: received with CorrelationId: 3d9297d8-5117-4bf3-8bb6-769b045ba68e.
[19:19:02.4539898]: InitialCommandReceived: received with CorrelationId: 8a02518f-c6f9-4fcb-a605-dede410e8610.
[19:19:02.4545988]: InitialCommandReceived: received with CorrelationId: 7773cd68-a58d-420f-bb6a-23fe67c5fcf1.
[19:19:02.5554364]: InitialCommandReceived: received with CorrelationId: aa12eb8a-fa9f-4b52-ae69-917ba04bdec3.
[19:19:02.6620557]: InitialCommandReceived: received with CorrelationId: b8ca11e2-cdcb-4388-8263-6e8df22f7cfb.
[19:19:02.7596565]: InitialCommandReceived: received with CorrelationId: 722e6aa7-4fb6-43c1-87a2-b09294e7f3c1.
--------------- 1 min -------------------------
[19:20:03.5855011]: InitialCommandReceived: received with CorrelationId: f1cdd00a-dd83-49bc-bf0d-33ab3052673c.
[19:20:03.5854979]: InitialCommandReceived: received with CorrelationId: da7c2aa7-c676-447d-8f5f-42d1e0546b7b.
[19:20:03.5856126]: InitialCommandReceived: received with CorrelationId: 264c9e84-b302-4511-a00a-6e6241e091e8.
[19:20:03.5860874]: InitialCommandReceived: received with CorrelationId: fc8f1f03-4103-4501-b406-88cc1c59f528.
[19:20:03.5893627]: InitialCommandReceived: received with CorrelationId: 779f42ae-8986-4fdb-b3c3-28113652717c.
[19:20:03.6894042]: InitialCommandReceived: received with CorrelationId: 7672425b-e0fc-4584-8a96-ff2d17689df8.
[19:20:03.8924069]: InitialCommandReceived: received with CorrelationId: 1182d62a-5657-4fb4-9877-efbc31155f01.
[19:20:03.9992448]: InitialCommandReceived: received with CorrelationId: 8fc8ecfe-5c14-4279-af71-95e48a1afa64.
--------------- 1 min -------------------------
[19:21:04.8731504]: InitialCommandReceived: received with CorrelationId: 3cf94c6a-2e5b-4740-9407-e12ae1ed9555.
[19:21:04.8732114]: InitialCommandReceived: received with CorrelationId: 478921d9-9a1a-49f1-a99d-2e5613625b74.
[19:21:05.2777634]: InitialCommandReceived: received with CorrelationId: 497d815f-466f-4baf-89cb-618e0bf2e9d3.
[19:21:05.2781711]: InitialCommandReceived: received with CorrelationId: d142326f-6b2a-4792-8cf8-0f0914aeea36.
[19:21:05.2782202]: InitialCommandReceived: received with CorrelationId: 718bce5a-c8e1-4163-a0df-ba795d8fc3d3.
[19:21:05.4816332]: InitialCommandReceived: received with CorrelationId: c9bfc039-7fdc-4cdf-b496-0e01d781b392.
[19:21:05.4816541]: InitialCommandReceived: received with CorrelationId: 4aa1a967-c9a3-4233-9d63-796b07af3aab.
[19:21:05.5837755]: InitialCommandReceived: received with CorrelationId: 7dfc6640-856a-4ef8-b5cc-c3469a6bb5bc.

核心原因 #

1分钟延迟的根源是Azure Service Bus会话的默认空闲超时设置：

• ASB标准层会话默认空闲超时为60秒，saga处理完成后，会话进入空闲状态，但链接会被保留至超时结束，期间无法复用该链接处理新的会话（对应新CorrelationId的消息）。
• 使用的MessageSessionRepository将saga状态绑定到会话生命周期，进一步延长了会话链接的占用时间，导致新消息需等待超时后才能被处理。

解决方案 #

1. 缩短会话空闲超时 #

在接收端点配置中调整SessionIdleTimeout，减少空闲会话的占用时间：

cfg.ReceiveEndpoint(queueAName, c =>
{
    c.RequiresSession = true;
    c.SessionIdleTimeout = TimeSpan.FromSeconds(5); // 自定义空闲超时时间
    c.ConfigureSaga<SimpleSagaState>(context);
});

会话空闲5秒后会自动关闭，释放链接资源，MassTransit可立即创建新会话处理下一批消息。

2. 更换持久化saga存储库（推荐） #

MessageSessionRepository仅适用于简单测试场景，生产环境建议使用外部持久化存储（如Azure Cosmos DB、SQL Server），摆脱会话生命周期的限制：

x.AddSagaStateMachine<SimpleSaga, SimpleSagaState>()
  .CosmosRepository(cosmos => 
  {
      cosmos.DatabaseEndpoint = "<你的Cosmos端点>";
      cosmos.DatabaseKey = "<你的Cosmos密钥>";
      cosmos.DatabaseName = "SagaDb";
      cosmos.ContainerName = "SimpleSagas";
  });

使用持久化存储后，会话仅用于消息路由，saga状态存储在外部数据库，消息处理完成后会话可立即关闭，无需等待超时。

3. 优化并发设置（可选） #

若需要更高吞吐量，可适当调高MaxConcurrentSessions值（需注意ASB标准层的并发连接配额限制）：

cfg.ReceiveEndpoint(queueAName, c =>
{
    c.RequiresSession = true;
    c.MaxConcurrentSessions = 16; // 根据实际业务调整
    c.SessionIdleTimeout = TimeSpan.FromSeconds(5);
    c.ConfigureSaga<SimpleSagaState>(context);
});

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