C# BackgroundService中如何正确并行运行含事务的任务？

首先，你当前的代码存在几个关键问题：

1. Parallel.For不支持异步委托：你在Parallel.For里用了async void的委托，这会导致DoWork的异步操作无法被正确等待，而且异步抛出的异常可能无法被捕获，属于潜在的bug。
2. 事务范围未启用异步流：默认的TransactionScope不支持异步操作的事务上下文流动，后续如果改用异步数据库操作，会直接脱离事务范围。
3. 未充分利用异步IO：数据库操作、文件读取都是IO密集型任务，用同步方法会阻塞线程，浪费系统资源。
4. 取消令牌未生效：stoppingToken没有传入到数据库操作、文件读取等关键步骤，服务停止时无法及时终止任务。

针对你的需求（持续利用可用线程执行DoWork，提升效率），推荐采用异步工作者循环+Task.WhenAll的模式，核心是利用异步IO释放线程，让系统高效调度任务，同时避免Parallel.For的异步兼容问题。

修改后的核心代码 #

ExecuteAsync方法（后台服务入口） #

protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
    _logger.LogInformation("Consume Scoped Service Hosted Service running.");

    // 工作者数量建议根据CPU核心数或系统负载调整，比如CPU核心数、数据库连接池大小
    var workerCount = Environment.ProcessorCount;
    var workerTasks = new List<Task>();

    for (int i = 0; i < workerCount; i++)
    {
        workerTasks.Add(Task.Run(async () =>
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    await DoWork(stoppingToken);
                    // 如果DoWork执行过快，可添加小延迟避免空循环占用CPU
                    // await Task.Delay(100, stoppingToken);
                }
                catch (OperationCanceledException)
                {
                    // 响应取消信号，退出循环
                    break;
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Worker task encountered an error");
                    // 出错后延迟重试，避免频繁报错
                    await Task.Delay(5000, stoppingToken);
                }
            }
        }, stoppingToken));
    }

    await Task.WhenAll(workerTasks);
}

DoWork方法（业务逻辑） #

private async Task DoWork(CancellationToken stoppingToken)
{
    _logger.LogInformation("Consume Scoped Service Hosted Service is working.");

    // 启用异步流，确保异步操作在事务范围内
    using var transactionScope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled);
    try
    {
        using var connection = new SqlConnection(connectionString);
        // 异步打开连接，传入取消令牌
        await connection.OpenAsync(stoppingToken);

        // 异步查询数据，注意原SQL缺少字段列表，这里补全*（实际建议指定具体字段）
        var query = "SELECT TOP(1) * FROM MyTable WITH (readpast, rowlock, updlock)";
        var row = (await connection.QueryAsync<MyType>(query, cancellationToken: stoppingToken)).FirstOrDefault();

        if (row == null)
        {
            // 无待处理数据，直接结束事务
            transactionScope.Complete();
            return;
        }

        // 异步读取文件，减少线程阻塞
        var fileContent = await File.ReadAllTextAsync(row.FileName, stoppingToken);
        var data = JsonSerializer.Deserialize<MyData>(fileContent);

        var item = new MyObject { Id = data.Id, Name = data.Name };
        
        // WCF无异步方法，只能同步调用，这里无法避免线程阻塞
        service.SendRequest(item);

        // 异步更新数据库，传入参数和取消令牌
        var updateQuery = "UPDATE MyTable SET Status = @Status WHERE Id = @Id";
        await connection.ExecuteAsync(updateQuery, 
            new { Status = "Processed", Id = row.Id }, 
            cancellationToken: stoppingToken);

        // 提交事务
        transactionScope.Complete();
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "Failed to execute DoWork");
        // using块会自动处理事务的回滚，无需手动调用Dispose
    }
}

方案优势 #

1. 异步友好：避免了Parallel.For的异步委托兼容问题，所有异步操作都能被正确等待和异常捕获。
2. 高效线程利用：异步IO操作（数据库、文件）会释放线程去处理其他任务，相比同步方法能支撑更高的并发。
3. 持续任务执行：每个工作者持续循环执行DoWork，直到收到停止信号，最大化利用系统资源。
4. 正确响应取消：取消令牌传入所有关键操作，服务停止时能及时终止任务，避免资源泄漏。

• 数据库连接池配置：确保连接字符串的Max Pool Size设置合理，避免并发过高导致连接耗尽。
• WCF客户端复用：不要每次创建新的WCF客户端，可复用ChannelFactory或线程安全的客户端实例，减少初始化开销。
• 并发数控制：workerCount不要盲目设置过大，需根据数据库、文件系统、WCF服务的承载能力调整，建议从CPU核心数开始测试。
• 事务超时：给TransactionScope设置超时时间（如new TransactionScope(TransactionScopeOption.Required, TimeSpan.FromMinutes(1), TransactionScopeAsyncFlowOption.Enabled)），避免长时间占用事务资源。
• 空数据处理：如果查询不到待处理数据，添加适当延迟再重试，避免空循环占用CPU。

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