Azure Service Bus会话队列并行处理消息：MaxConcurrentCallsPerSession不生效

问题描述 #

我尝试在启用会话的Azure Service Bus队列中并行处理多条消息，已将MaxConcurrentCallsPerSession设置为5，但仍只能逐个接收消息。使用.NET Framework 4.7.2和Azure.Messaging.ServiceBus 7.17.4编写的控制台应用示例代码中，期望并行接收并完成消息，实际却为串行处理。请问该需求是否可实现？

示例代码 #

static void Main()
{
    MainAsync().Wait();
}

static async Task MainAsync()
{
    //create the queue
    await CreateQueue();

    //initialize queue client
    ServiceBusClient queueClient = new ServiceBusClient(_serviceBusConnectionString, new ServiceBusClientOptions
    {
        TransportType = ServiceBusTransportType.AmqpWebSockets,
    });

    //initialize the sender
    ServiceBusSender sender = queueClient.CreateSender(_queueName);

    //queue 3 messages
    await sender.SendMessageAsync(new ServiceBusMessage() { SessionId = _sessionId, MessageId = "1" });
    await sender.SendMessageAsync(new ServiceBusMessage() { SessionId = _sessionId, MessageId = "2" });
    await sender.SendMessageAsync(new ServiceBusMessage() { SessionId = _sessionId, MessageId = "3" });

    //initialize processor
    ServiceBusSessionProcessor processor = queueClient.CreateSessionProcessor(_queueName, new ServiceBusSessionProcessorOptions()
    {
        AutoCompleteMessages = false,
        ReceiveMode = ServiceBusReceiveMode.PeekLock,
        SessionIds = { _sessionId },
        PrefetchCount = 5,
        MaxConcurrentCallsPerSession = 5
    });

    //add message handler
    processor.ProcessMessageAsync += HandleReceivedMessage;

    //add error handler
    processor.ProcessErrorAsync += ErrorHandler;

    //start the processor
    await processor.StartProcessingAsync();

    Console.ReadLine();
}

static async Task CreateQueue()
{
    ServiceBusAdministrationClient client = new ServiceBusAdministrationClient(_serviceBusConnectionString);

    bool doesQueueExist = await client.QueueExistsAsync(_queueName);

    //check if the queue exists, if not then create one
    if (!doesQueueExist)
    {
        _ = await client.CreateQueueAsync(new CreateQueueOptions(_queueName)
        {
            RequiresSession = true,
            DeadLetteringOnMessageExpiration = true,
            MaxDeliveryCount = 3,
            EnableBatchedOperations = true,
        });
    }
}

static async Task HandleReceivedMessage(ProcessSessionMessageEventArgs sessionMessage)
{
    Console.WriteLine("Received message: " + sessionMessage.Message.MessageId);

    await Task.Delay(5000).ConfigureAwait(false);

    await sessionMessage.CompleteMessageAsync(sessionMessage.Message);

    Console.WriteLine("Completed message: " + sessionMessage.Message.MessageId);
}

static Task ErrorHandler(ProcessErrorEventArgs e)
{
    Console.WriteLine("Error received");

    return Task.CompletedTask;
}

期望输出 #

Received message: 1
Received message: 2
Received message: 3
Completed message: 1
Completed message: 2
Completed message: 3

实际输出 #

Received message: 1
Completed message: 1
Received message: 2
Completed message: 2
Received message: 3
Completed message: 3

解答 #

需求可行性 #

该需求可以实现，但需注意：

• Azure Service Bus会话的核心设计是保证同一会话内消息的顺序处理，启用同一会话内消息并行处理会打破这一顺序保证。
• MaxConcurrentCallsPerSession参数的作用正是控制单一会话内可并发处理的消息数量，将其设置为大于1的值（如示例中的5）即可实现同一会话内的并行处理。

为何示例代码未生效 #

以下是可能的原因及排查方向：

1. 使用Azure Service Bus模拟器：
   模拟器可能不支持同一会话内的并行处理特性，建议切换至云环境的Service Bus namespace测试。
2. 消息发送与处理器启动的时序问题：
   虽然示例中已通过await确保消息发送完成后再启动处理器，但极端情况下可能存在消息未完全进入队列的延迟。可在启动处理器前添加短暂延迟（如await Task.Delay(1000)）验证。
3. 版本兼容性：
   确认使用的Azure.Messaging.ServiceBus版本（7.17.4）确实支持MaxConcurrentCallsPerSession参数（该参数自7.11.0版本引入，7.17.4符合要求）。
4. 会话锁相关配置：
   检查队列的会话锁持续时间是否过短（默认60秒，示例中5秒延迟远小于该值，此可能性较低）。

注意事项 #

• 并行处理同一会话内消息会丢失顺序保证，仅适用于不依赖消息顺序的场景。
• 若需保持顺序同时提升处理效率，可将消息分配至不同会话（使用不同SessionId），通过MaxConcurrentSessions参数控制多会话的并行处理。

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