.NET Core 3.1读取Service Bus队列指定数量消息的优化咨询

原代码通过注册消息回调加轮询等待的方式读取消息，存在线程安全风险、逻辑繁琐且未正确处理消息完成的问题。以下是两种更简洁可靠的实现方案：

方案一：基于旧SDK（Microsoft.Azure.ServiceBus）的优化实现 #

直接使用ReceiveAsync批量接收消息，线性化逻辑，避免多线程回调的复杂性：

static async Task<List<string>> ReceiveMessagesAsync(int messageCount, TimeSpan timeout)
{
    var messages = new List<string>();
    // 使用using自动管理QueueClient生命周期，无需手动Close
    using var queueClient = new QueueClient(ServiceBusConnectionString, QueueName);

    try
    {
        int remainingMessages = messageCount;
        var cancellationTokenSource = new CancellationTokenSource(timeout);

        while (remainingMessages > 0 && !cancellationTokenSource.Token.IsCancellationRequested)
        {
            // 批量接收剩余数量的消息，减少IO调用次数
            var receivedMessages = await queueClient.ReceiveAsync(remainingMessages, timeout);
            if (receivedMessages == null || receivedMessages.Count == 0)
            {
                break; // 无更多消息，退出循环
            }

            foreach (var message in receivedMessages)
            {
                var json = Encoding.UTF8.GetString(message.Body);
                messages.Add(json);
                
                // 标记消息为已完成，防止锁过期后重新入队
                await queueClient.CompleteAsync(message.SystemProperties.LockToken);
                remainingMessages--;
            }
        }
    }
    catch (OperationCanceledException)
    {
        // 超时取消，属于正常流程
    }
    catch (Exception ex)
    {
        Console.WriteLine($"接收消息出错: {ex.Message}");
    }

    return messages;
}

该方案的优势： #

• 逻辑线性直观，无需依赖回调和轮询等待
• 自带超时控制，无需手动计算时间差判断无消息
• 自动管理客户端生命周期，避免资源泄漏
• 线程安全，所有操作在同一异步流中执行
• 正确完成消息，避免重复消费

方案二：基于最新SDK（Azure.Messaging.ServiceBus）的实现 #

旧版SDK已被弃用，微软推荐使用Azure.Messaging.ServiceBus，API设计更简洁现代：

using Azure.Messaging.ServiceBus;

static async Task<List<string>> ReceiveMessagesAsync(int messageCount, TimeSpan timeout)
{
    var messages = new List<string>();
    await using var client = new ServiceBusClient(ServiceBusConnectionString);
    await using var receiver = client.CreateReceiver(QueueName);

    try
    {
        int remainingMessages = messageCount;
        var cancellationTokenSource = new CancellationTokenSource(timeout);

        while (remainingMessages > 0 && !cancellationTokenSource.Token.IsCancellationRequested)
        {
            var receivedMessages = await receiver.ReceiveMessagesAsync(remainingMessages, timeout, cancellationTokenSource.Token);
            if (receivedMessages.Count == 0)
            {
                break;
            }

            foreach (var message in receivedMessages)
            {
                var json = Encoding.UTF8.GetString(message.Body);
                messages.Add(json);
                
                await receiver.CompleteMessageAsync(message, cancellationTokenSource.Token);
                remainingMessages--;
            }
        }
    }
    catch (OperationCanceledException)
    {
        // 超时处理
    }
    catch (Exception ex)
    {
        Console.WriteLine($"接收消息出错: {ex.Message}");
    }

    return messages;
}

新SDK的额外优势： #

• API更清晰，拆分ServiceBusClient和ServiceBusReceiver职责
• 更好的异步取消令牌集成
• 性能优化更出色，支持更多高级特性

原代码的核心问题说明 #

1. 线程安全风险：messages列表在回调线程和主线程同时修改，未加同步锁，可能引发异常
2. 未处理消息完成：仅读取消息但未调用CompleteAsync，消息会在锁过期后重新入队，导致重复消费
3. 客户端生命周期管理不当：类级别queueClient可能被重复使用，未确保每次使用后正确释放
4. 超时逻辑繁琐：手动计算时间差判断无消息，代码冗余且易出错

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