Azure Service Bus多订阅监听方案验证及向MessageHandler传参问题

1. 现有并行监听方案的合理性分析 #

现有方案不合理，存在3个核心问题：

• Parallel.ForEach不支持异步委托：你传入的async lambda会被识别为async void，循环不会等待异步操作完成就会直接退出，处理器可能还没完成初始化就被提前释放，服务稳定性完全无法保障。
• 连接资源浪费：官方建议相同连接字符串的ServiceBusClient要复用，你每个job都新建实例，会额外创建大量不必要的AMQP连接，不仅拉高资源开销，还可能触发服务端连接数限制。
• 控制台输入逻辑错误：Console.ReadKey放在并行循环内部，会导致多个线程同时争抢控制台输入，你按一次按键只会停止其中某一个监听任务，其余任务会继续运行，完全不符合全局停止的预期逻辑。

合理的实现思路是：改用同步遍历创建所有处理器，主线程统一等待停止信号，最后批量释放所有资源。

2. 给MessageHandler传递job对象的解决方案 #

直接用闭包捕获当前遍历的job对象即可，不需要用静态处理方法，修正后的完整代码如下：

var jobs = GetAllServiceBusTopics();
// 缓存同连接字符串的ServiceBusClient，避免重复创建
var clientCache = new Dictionary<string, ServiceBusClient>();
// 保存所有处理器，后续统一停止释放
var processors = new List<ServiceBusProcessor>();

try
{
    foreach (var job in jobs)
    {
        var connString = job.Environment.ServiceBusConnectionString;
        // 复用同连接的Client
        if (!clientCache.TryGetValue(connString, out var client))
        {
            client = new ServiceBusClient(connString);
            clientCache[connString] = client;
        }

        var processor = client.CreateProcessor(job.Environment.TopicName, _subscriptionName, new ServiceBusProcessorOptions());
        
        // 用闭包捕获当前job，不需要额外传参
        processor.ProcessMessageAsync += async args =>
        {
            // 这里可以直接使用当前job的所有属性
            Console.WriteLine($"处理来自{job.Environment.TopicName}的消息：{args.Message.MessageId}");
            await args.CompleteMessageAsync(args.Message);
        };
        // 错误处理也可以通过闭包传递job
        processor.ProcessErrorAsync += async args =>
        {
            Console.WriteLine($"订阅{job.Environment.TopicName}出现错误：{args.Exception.Message}");
            await Task.CompletedTask;
        };

        await processor.StartProcessingAsync();
        processors.Add(processor);
    }

    Console.WriteLine("所有监听已启动，按任意键停止");
    Console.ReadKey();

    // 统一停止所有处理器
    foreach (var processor in processors)
    {
        await processor.StopProcessingAsync();
    }
}
finally
{
    // 统一释放所有资源
    foreach (var processor in processors)
    {
        await processor.DisposeAsync();
    }
    foreach (var client in clientCache.Values)
    {
        await client.DisposeAsync();
    }
}

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