托管服务中如何实现多队列并行出队互不阻塞且单队列顺序执行？

你需要的按表隔离队列、多队列并行、单队列串行的调度逻辑，可以用.NET 原生的Channel<T> + 单队列专属消费协程的方案实现，不需要手动处理复杂的线程同步逻辑，性能和稳定性都有保障。

核心实现步骤 #

• 用线程安全的Channel<string>替换你当前使用的非线程安全Queue，避免多线程读写队列的并发问题
• 为每个表的Channel启动一个独立的长时间运行的消费协程，协程内部循环拉取作业逐个执行，天然保证单队列内作业的执行顺序
• 作业生产端只需要按表名找到对应Channel，写入作业ID即可，写入操作原生支持多线程并发调用

参考代码 #

using System.Collections.Concurrent;
using System.Threading.Channels;
using Microsoft.Extensions.Hosting;

public class TableJobProcessingService : BackgroundService
{
    // 存储每个表对应的Channel队列，ConcurrentDictionary线程安全，支持动态增删表队列
    private readonly ConcurrentDictionary<string, Channel<string>> _tableChannels = new();
    private readonly List<string> _initialTableNames = new() { "Table1", "Table2", "Table100" };

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // 初始化预设表的队列和消费协程
        foreach (var tableName in _initialTableNames)
        {
            await InitTableChannelAsync(tableName, stoppingToken);
        }

        // 阻塞等待服务停止信号
        await Task.Delay(Timeout.Infinite, stoppingToken);
    }

    // 初始化单个表的队列和消费协程
    private async Task InitTableChannelAsync(string tableName, CancellationToken stoppingToken)
    {
        var channel = Channel.CreateUnbounded<string>(); // 也可以用CreateBounded控制队列最大长度做背压
        _tableChannels.TryAdd(tableName, channel);

        // 启动当前表专属的消费协程
        _ = ConsumeTableJobAsync(tableName, channel.Reader, stoppingToken);
        await Task.CompletedTask;
    }

    // 单个表的消费逻辑，每个表独立一个该协程实例
    private async Task ConsumeTableJobAsync(string tableName, ChannelReader<string> reader, CancellationToken stoppingToken)
    {
        // 循环拉取队列中的作业，队列空时会自动阻塞等待新作业，不会占用CPU资源
        await foreach (var jobId in reader.ReadAllAsync(stoppingToken))
        {
            try
            {
                // 这里写你实际的作业执行逻辑，执行完成后再进入下一次循环拿下一个作业，保证单队列串行
                Console.WriteLine($"开始处理表{tableName}的作业{jobId}");
                // await ProcessJobAsync(jobId); 
            }
            catch (Exception ex)
            {
                // 单个作业执行失败的处理逻辑，比如重试、打日志、告警，不影响同队列后续作业和其他队列
                Console.WriteLine($"表{tableName}的作业{jobId}执行失败：{ex.Message}");
            }
        }
    }

    // 作业生产端调用的写入接口，你说的另外一个定时加载作业的托管服务直接调用这个方法即可
    public async Task EnqueueJobAsync(string tableName, string jobId)
    {
        if (!_tableChannels.TryGetValue(tableName, out var channel))
        {
            // 如果是新增的表，动态创建队列和消费协程，不需要重启服务
            var stoppingToken = new CancellationTokenSource().Token; // 实际使用时可以注入全局停止Token
            await InitTableChannelAsync(tableName, stoppingToken);
            channel = _tableChannels[tableName];
        }
        await channel.Writer.WriteAsync(jobId);
    }
}

// 主函数注册服务
public class Program
{
    static void Main(string[] args)
    {
        var host = Host.CreateDefaultBuilder(args)
            .AddHostedService<TableJobProcessingService>()
            .Build();
        
        host.Run();
    }
}

效果说明 #

每个表的消费协程完全独立运行，互不阻塞：小表的作业执行快，消费协程会快速拉取下一个作业执行；大表的作业执行慢，只会阻塞自己队列的后续作业，不会影响其他表的作业调度，完全符合你要求的Table2完成1个作业的时长内Table1可完成10个作业的效果。

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