.NET Core实现Azure Service Bus Queue持续监听新消息方法

安全阿岳

2026-8-28

实现方案说明

Azure.Messaging.ServiceBus 命名空间下提供了官方内置的持续消息消费组件 ServiceBusProcessor，无需自行编写while无限循环或自定义长运行Task实现拉取逻辑。该组件内部已经封装了长轮询、连接故障自动恢复、消息锁自动续期、并发流控、异常重试等底层能力，可同时处理监听启动前的存量消息、以及启动后新入队的持续消息流，不会出现存量消费完成后无法接收新消息的问题。

你之前遇到的消费断流问题，基本都是因为使用了单次/批量ReceiveMessageAsync手动拉取模式、没有持续触发拉取逻辑，或是客户端/处理器实例被提前释放导致的。

具体实现步骤

第一步：安装依赖包
引入NuGet包 Azure.Messaging.ServiceBus，不要使用旧版Microsoft.Azure.ServiceBus包，旧包已停止迭代。
第二步：初始化客户端与处理器
以控制台应用/.NET Core通用主机场景为例，核心代码如下：

// 初始化Service Bus客户端，建议全局单例复用
var serviceBusClient = new ServiceBusClient("<你的Service Bus命名空间连接字符串>");

// 配置处理器参数
var processorOptions = new ServiceBusProcessorOptions
{
    MaxConcurrentCalls = 8, // 单实例最大并发处理消息数，根据业务承载能力调整
    AutoCompleteMessages = false, // 生产环境建议关闭自动确认，由业务代码手动控制消息状态
    MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(15), // 自动续消息锁的最长时长，需大于业务单条消息最长处理时间
    ReceiveMode = ServiceBusReceiveMode.PeekLock // 默认为 peek-lock 模式，保证消息至少被消费一次
};

// 创建指定队列的处理器实例
var queueProcessor = serviceBusClient.CreateProcessor("<目标队列名称>", processorOptions);

第三步：注册处理回调

// 注册消息处理回调：只要有可消费的消息（存量/新入队）就会触发该逻辑
queueProcessor.ProcessMessageAsync += async (processArgs) =>
{
    var message = processArgs.Message;
    var messageContent = message.Body.ToString();
    try
    {
        // 编写你的业务处理逻辑
        Console.WriteLine($"消费消息ID：{message.MessageId}，内容：{messageContent}");

        // 业务处理成功后，标记消息为完成状态，Service Bus会从队列中移除该消息
        await processArgs.CompleteMessageAsync(message);
    }
    catch (Exception ex)
    {
        // 业务处理失败时，可选择放弃消息，让消息重新投递供其他消费者消费
        await processArgs.AbandonMessageAsync(message);
        // 若消息重试多次仍失败，可转入死信队列：await processArgs.DeadLetterMessageAsync(message, ex.Message);
    }
};

// 注册异常处理回调：拉取/处理过程中出现的连接异常、传输异常都会触发该事件，处理器会自动做故障恢复
queueProcessor.ProcessErrorAsync += (errorArgs) =>
{
    Console.WriteLine($"消费异常来源：{errorArgs.ErrorSource}，异常信息：{errorArgs.Exception.Message}");
    return Task.CompletedTask;
};

第四步：启动监听

// 启动处理器，该方法调用后会开始持续拉取消息，不会主动退出
await queueProcessor.StartProcessingAsync();

// 应用停止时优雅关闭处理器即可
// await queueProcessor.StopProcessingAsync();
// await serviceBusClient.DisposeAsync();

注意事项

ServiceBusClient和ServiceBusProcessor实例建议在应用生命周期内单例复用，不要放在using块中随短生命周期作用域释放，否则会出现消费中断的问题。
不需要自行编写任何循环拉取逻辑，ServiceBusProcessor内部已经实现了生产级的持续监听机制，存量消息消费完成后会保持长连接等待新消息，新消息入队后会立刻触发处理回调，无消费断流问题。
如果是在ASP.NET Core等带通用主机的项目中使用，可以把上述初始化、启动、停止逻辑封装在IHostedService中，跟随应用生命周期托管运行即可。

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