如何强制停止ProcessMessageAsync？Function App Service Bus线程残留问题

场景与代码实现 #

Function App需求为每5分钟调用Service Bus收集消息，5秒后取消处理以响应消费者，原实现代码如下：

// the client that owns the connection and can be used to create senders and receivers
client = new ServiceBusClient(connectionString);

// the processor that reads and processes messages from the queue
processor = client.CreateProcessor(topicName, subscriptionName, new ServiceBusProcessorOptions());

// handler to process SB message async
processor.ProcessMessageAsync += MessageHandler;

// handle in receiving SB messages 
async Task MessageHandler(ProcessMessageEventArgs args)
{
            string body = args.Message.Body.ToString();
            var message = JsonConvert.DeserializeObject<JObject>(body);
            Messages.Add(message);

            // complete the message. Message is deleted from the queue
            await args.CompleteMessageAsync(args.Message);
}

// handle any errors when receiving messages
Task ErrorHandler(ProcessErrorEventArgs args)
{
            Console.WriteLine(args.Exception.ToString());
            var errorObject = new JObject();
            errorObject["ErrorMessage"] = $"Error processing the request from the queue: {args.Exception.Message}";
            Messages.Add(errorObject);
            return Task.CompletedTask;
}

// handler to process any errors
processor.ProcessErrorAsync += ErrorHandler;

// cancel the token after 5 secs
tokenSource.CancelAfter(TimeSpan.FromSeconds(5));

// start process async
await processor.StartProcessingAsync(tokenSource.Token);

// end the while loop for any cancelation process (5 secs)
while (!tokenSource.IsCancellationRequested) {}

// stop or end process async
await processor.StopProcessingAsync(tokenSource.Token);

问题现象 #

当客户端/Postman多次取消Function App调用后，虽然调用本身显示已取消，但Function App线程池中会残留Service Bus处理器实例。后续向Service Bus队列发送消息时，这些残留的处理器会自动拾取消息，导致正常调用Function App时显示无消息可用。

模拟步骤 #

• 调用Function App并在处理过程中取消操作（重复3次）
• 向Service Bus发送消息，消息会被残留的处理器自动拾取

临时解决方法：重启Function App

根本原因与修复方案 #

根本原因 #

原代码未正确释放ServiceBusClient和ServiceBusProcessor资源，每次函数调用创建的实例在取消后未被妥善清理，导致线程池中残留的实例持续监听Service Bus队列。

修复方案 #

1. 用using语句强制管理资源生命周期
   ServiceBusClient和ServiceBusProcessor都实现了IDisposable接口，必须确保函数结束时释放资源，避免实例残留。

2. 优化取消逻辑
   移除空的while循环，改用Task.Delay配合取消令牌实现等待，同时通过try/finally块确保处理器一定会停止。

修复后的完整代码：

using var client = new ServiceBusClient(connectionString);
using var processor = client.CreateProcessor(topicName, subscriptionName, new ServiceBusProcessorOptions());

processor.ProcessMessageAsync += MessageHandler;
processor.ProcessErrorAsync += ErrorHandler;

var tokenSource = new CancellationTokenSource();
tokenSource.CancelAfter(TimeSpan.FromSeconds(5));

try
{
    await processor.StartProcessingAsync(tokenSource.Token);
    // 等待取消信号，替代空while循环
    await Task.Delay(Timeout.Infinite, tokenSource.Token);
}
catch (OperationCanceledException)
{
    // 捕获取消异常，无需额外处理
}
finally
{
    // 确保处理器停止，不受取消令牌影响
    await processor.StopProcessingAsync(CancellationToken.None);
}

async Task MessageHandler(ProcessMessageEventArgs args)
{
    string body = args.Message.Body.ToString();
    var message = JsonConvert.DeserializeObject<JObject>(body);
    Messages.Add(message);

    await args.CompleteMessageAsync(args.Message);
}

Task ErrorHandler(ProcessErrorEventArgs args)
{
    Console.WriteLine(args.Exception.ToString());
    var errorObject = new JObject();
    errorObject["ErrorMessage"] = $"Error processing the request from the queue: {args.Exception.Message}";
    Messages.Add(errorObject);
    return Task.CompletedTask;
}

3. 关键注意点

• ServiceBusClient是重量级对象，若需复用可考虑单例模式，但在函数每次调用创建的场景下，必须用using释放；
• 取消令牌需传递到所有响应取消的方法中，确保处理流程能及时终止；
• finally块保证无论是否发生异常，处理器都会停止，彻底清理监听逻辑。

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