C# TPL如何平均分配记录给有限任务并行处理并实时返回结果

不要手动拆分记录列表给固定数量的Task——这种方式很容易出现负载不均（部分Task提前跑完空闲、部分Task积压慢请求），容错性也差。最稳妥的实现是采用受限并发的工作队列模式，既可以严格卡死并发上限，又能支持单条记录处理完实时返回结果，资源利用率最高。

推荐方案：使用原生Parallel.ForEachAsync（.NET 6+ 支持） #

这是.NET官方提供的并行处理API，不需要任何第三方依赖，核心优势：

• 通过MaxDegreeOfParallelism参数硬限制最大并发数，全程不会超量
• 自动做任务调度，哪个工作线程/Task空闲就分配下一条待处理记录，不会出现忙闲不均
• 单条记录处理逻辑独立，跑完即可立刻更新状态，不需要等待其他记录
• 内置取消令牌支持，需要中途终止批量处理时可以随时停掉所有任务

完整实现代码如下，已经整合你现有的API调用、DB写入逻辑，以及实时状态更新的要求：

// 👉 这个值根据下游API、SQL库的压测结果调整，是全局并发硬上限
int maxConcurrentTaskCount = 8;

// 拉取待处理的全量记录
List<ShippingConfig> pendingRecords = await configRepository.GetPendingShippingConfigs();

// 配置并行规则
var parallelOptions = new ParallelOptions
{
    MaxDegreeOfParallelism = maxConcurrentTaskCount,
    TaskScheduler = TaskScheduler.Default // 服务端场景用默认调度器即可
};

// 启动并行处理
await Parallel.ForEachAsync(pendingRecords, parallelOptions, async (record, cancellationToken) =>
{
    try
    {
        // 标记为处理中，实时落库，调用方可以立刻看到该记录进入处理流程
        record.Processed = "N";
        await configRepository.UpdateRecordProcessStatus(record);

        // 构造API请求参数
        var apiRequest = BuildPublishRequest(record);
        // 原有API调用逻辑
        var pricingResult = await shippingService.PublishConfigShipping(apiRequest, cancellationToken);
        
        if (pricingResult.Item2 == System.Net.HttpStatusCode.OK)
        {
            // 原有DB保存逻辑
            await configRepository.SaveConfigShipping(record, cancellationToken);
            // 标记处理成功
            record.Processed = "Y";
            record.Status = "Pass";
        }
        else
        {
            // 接口返回非200状态，标记失败
            record.Processed = "Y";
            record.Status = "Failed";
            record.Remark = $"API返回异常：{pricingResult.Item1}";
        }
    }
    catch (Exception ex)
    {
        // 捕获单条记录的所有异常（网络闪断、DB超时、参数错误等），避免单条失败终止整个批量流程
        record.Processed = "Y";
        record.Status = "Failed";
        record.Remark = $"处理异常：{ex.Message}";
    }
    finally
    {
        // 无论成功失败，实时更新最终处理状态到DB，调用方不需要等全量跑完就能查到结果
        await configRepository.UpdateRecordProcessStatus(record, cancellationToken);
        // 如果需要主动推送给前端，可以在这里调用SignalR/消息队列发送单条处理结果
    }
});

低版本.NET兼容方案：使用Channel实现工作队列 #

如果你的项目用的是.NET 5及以下版本，没有内置Parallel.ForEachAsync，可以用System.Threading.Channels实现完全一致的效果：启动固定数量的常驻工作Task，从通道中读取待处理记录执行，全程并发数不会超过工作Task的数量。

int maxConcurrentTaskCount = 8;
List<ShippingConfig> pendingRecords = await configRepository.GetPendingShippingConfigs();

// 初始化有界通道
var channel = Channel.CreateBounded<ShippingConfig>(new BoundedChannelOptions(pendingRecords.Count)
{
    FullMode = BoundedChannelFullMode.Wait
});

// 启动固定数量的工作任务
var workTasks = Enumerable.Range(0, maxConcurrentTaskCount)
    .Select(async _ =>
    {
        await foreach (var record in channel.Reader.ReadAllAsync())
        {
            // 单条记录的处理逻辑和上面ForEachAsync中的逻辑完全一致，包含状态更新、异常捕获
            await ProcessSingleShippingRecord(record);
        }
    })
    .ToArray();

// 把所有待处理记录写入通道
foreach (var record in pendingRecords)
{
    await channel.Writer.WriteAsync(record);
}
channel.Writer.Complete();

// 等待所有记录处理完成
await Task.WhenAll(workTasks);

关键注意事项 #

• 不要手动拆分列表给固定Task：比如把1000条记录平均拆成8份开8个Task跑，一旦某份记录全是慢请求，其他Task提前跑完也会处于空闲状态，整体处理耗时会变长，而且单个Task崩溃会导致整块记录丢失，容错性极差。
• 并发数不要拍脑袋设置：可以先按「下游API允许的峰值QPS × 单条请求平均耗时」估算初始值，比如API允许100QPS，单条请求平均耗时80ms，初始并发设为8即可，上线前一定要做压测，观察API错误率、DB的CPU/磁盘IO负载，逐步调整到最优值。
• 不要用Task.WhenAll+SemaphoreSlim一次性创建所有Task：这种方式会把所有记录对应的Task对象一次性创建在内存中，记录量很大的时候会造成不必要的内存占用，调度效率远低于上面两种按需取任务的方案。
• 给所有外部调用加超时和取消支持：API请求、DB操作都要配置合理的超时时间，避免某条记录卡死长期占用并发槽位，拖慢整个批量处理流程。
• 如果DB写入性能是瓶颈，可以单独给DB操作加一层并发限制（比如用独立的SemaphoreSlim控制DB写入并发为3，API调用并发保持8），灵活匹配不同下游服务的承载能力。

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