Azure Service Bus Client：如何在C#中并发处理消息

问题 #

对接Azure Service Bus接收数据提供商的消息时，存在单条消息处理阻塞问题。核心代码如下，调大MaxConcurrentCalls至2或4后并发处理仍未生效，部分消息处理耗时超10分钟会阻塞其他消息：

var _serviceBusClient = new ServiceBusClient(_options.FullyQualifiedNamespace, token);

var options = new ServiceBusProcessorOptions
{
    AutoCompleteMessages = false,
    MaxConcurrentCalls = 1,
    PrefetchCount = 1,
    MaxAutoLockRenewalDuration = new TimeSpan(0, 0, 20, 0),
};

_serviceBusProcessor = _serviceBusClient.CreateProcessor(queueName, options);

_serviceBusProcessor.ProcessMessageAsync += messageHandler.DispatchAsync;
_serviceBusProcessor.ProcessErrorAsync += messageHandler.ErrorHandlerAsync;

......

public class MessageHandler : IMessageHandler
{
    public async Task DispatchAsync(ProcessMessageEventArgs args)
    {
        // 消息处理逻辑，部分场景耗时约10分钟
        if (args.Message.Subject == "") { TransformData(args.Message.Body); }
        ......
        await args.CompleteMessageAsync(args.Message);
    }
}

解决方案 #

1. 调整处理器核心配置 #

• 将MaxConcurrentCalls设置为目标并发数（如4），同时把PrefetchCount调整为MaxConcurrentCalls的2-3倍，确保预取足够的消息供并发处理，避免线程空等：

var options = new ServiceBusProcessorOptions
{
    AutoCompleteMessages = false,
    MaxConcurrentCalls = 4,
    PrefetchCount = 8, // 建议值：MaxConcurrentCalls * 2
    MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(20),
};

2. 确保消息处理逻辑真正异步 #

之前调大MaxConcurrentCalls无效的核心原因，大概率是DispatchAsync中存在同步阻塞代码（比如TransformData是同步方法），占用线程池线程导致无法启动新的并发任务。需做以下调整：

• 如果TransformData是IO密集型操作，替换为异步版本并使用await调用；
• 如果是CPU密集型操作，用Task.Run包装同步逻辑，释放当前线程供SDK启动其他并发任务：

public async Task DispatchAsync(ProcessMessageEventArgs args)
{
    if (args.Message.Subject == "") 
    {
        // CPU密集场景：用Task.Run包装同步方法
        await Task.Run(() => TransformData(args.Message.Body));
        // IO密集场景：优先使用异步方法
        // await TransformDataAsync(args.Message.Body);
    }
    // 其他业务逻辑
    await args.CompleteMessageAsync(args.Message);
}

3. 验证锁续期有效性 #

当前MaxAutoLockRenewalDuration设置为20分钟，足够覆盖10分钟的处理时长，但需确保SDK能正常自动续期消息锁，避免消息因锁过期重回队列。若处理耗时可能超过20分钟，需适当延长该值。

内容的提问来源于stack exchange，提问作者ZZzz.George