C#中多线程异步操作的方法-Parallel.ForEachAsync-平芜编程栈

1. 核心概念

Parallel.ForEachAsync是在.NET 6中引入的方法。
它的核心作用是：对一组数据执行异步操作，并且同时控制并发数量（也就是同时跑多少个任务）。
它完美解决了两个痛点：

传统的Parallel.ForEach不支持async/await（它是为阻塞式 CPU 任务设计的）。
直接用Task.WhenAll(tasks)会一次性把所有任务扔出去，如果数据有 10 万条，会瞬间耗尽内存或网络端口。

2. 基本语法与关键参数

publicstaticTaskForEachAsync<TSource>(thisIEnumerable<TSource>source,ParallelOptionsparallelOptions,Func<TSource,CancellationToken,ValueTask>body)

关键参数说明：

source：数据源（如数组、List、或你用Chunk()切分出的块）。
parallelOptions：并行配置项，极其重要，包含两个核心属性：
- MaxDegreeOfParallelism：最大并发数。比如设为 5，表示最多同时运行 5 个异步任务，完成一个再补一个。
- CancellationToken：取消令牌，用于中途终止所有并发任务。
body：要执行的异步逻辑（注意返回类型是ValueTask，不是Task，性能更好）。

3. 经典示例

示例 1：批量并发下载网页/API请求（最常见场景）

假设我们要下载 100 个网页，如果串行可能需要 1 分钟，但如果同时开 10 个并发，只需 6 秒。

usingSystem.Net.Http;List<string>urls=GetUrls();// 假设有 100 个 URLusingHttpClientclient=newHttpClient();// 配置并行选项：最大并发 10varoptions=newParallelOptions{MaxDegreeOfParallelism=10};Console.WriteLine("开始并发下载...");awaitParallel.ForEachAsync(urls,options,async(url,ct)=>{try{// 传入 ct，如果外部取消，当前请求会立即中止stringcontent=awaitclient.GetStringAsync(url,ct);Console.WriteLine($"成功下载{url}，长度:{content.Length}");}catch(OperationCanceledException){// 被取消}catch(Exceptionex){Console.WriteLine($"下载失败{url}:{ex.Message}");}});Console.WriteLine("全部下载完成。");

示例二：收集结果

ForEachAsync 本身没有返回值（返回 Task），如果需要收集处理结果，必须使用线程安全集合（如 ConcurrentBag 或 ConcurrentQueue）。**

usingSystem.Collections.Concurrent;usingSystem.Diagnostics;publicstaticasyncTaskExample_WithResultCollection(){varinputs=Enumerable.Range(1,10);varresults=newConcurrentBag<int>();// 线程安全集合awaitParallel.ForEachAsync(inputs,asynci=>{awaitTask.Delay(100);// 模拟耗时操作intresult=i*i;// 计算逻辑results.Add(result);// 并发添加结果（线程安全）Console.WriteLine($"{DateTime.Now:mm:ss:fff}- 处理{i}");});Console.WriteLine($"结果数量:{results.Count}");}

示例三：配合取消令牌

在用户关闭窗口或点击“停止”按钮时优雅地停止处理。

publicstaticasyncTaskExample_WithCancellation(CancellationTokencts){varrange=Enumerable.Range(0,1000);try{awaitParallel.ForEachAsync(range,newParallelOptions{CancellationToken=cts,MaxDegreeOfParallelism=10},async(i,ct)=>{// 如果外部取消，这里会抛出 OperationCanceledExceptionawaitTask.Delay(100,ct);Console.WriteLine($"处理中:{i}");});}catch(OperationCanceledException){Console.WriteLine("任务被用户取消。");}}

4. 核心优势对比表

特性	`foreach`+`await`	`Task.WhenAll`	`Parallel.ForEachAsync`
执行方式	串行（一个接一个）	全量并发（同时启动所有）	可控并发（池化执行）
内存/资源占用	极低	极高（容易内存溢出/端口耗尽）	适中（按需启动）
总耗时	最长	最短（如果不崩溃的话）	较短（与并发数成正比）
支持取消	支持	支持	完美支持

5. 注意事项与避坑指南

必须限制并发数：
不设置MaxDegreeOfParallelism时，默认可能会使用处理器数量作为并发度。对于网络请求，建议显式设置为5到20之间（根据目标服务器的承受能力和你的网络带宽决定），防止把对方服务器打挂，或触发对方防爬机制被封 IP。
线程安全问题（最容易踩的坑）：
在并发环境下，Parallel.ForEachAsync的内部相当于多线程同时运行。如果你在里面修改了 UI 绑定的集合（如ObservableCollection）或普通的计数字典，极大概率会报错或数据错乱。
- 集合操作：使用ConcurrentDictionary或ConcurrentBag，或者用lock。
- UI 更新：如果要在里面更新 WPF/WinForms 的 UI，需要使用Dispatcher.Invoke。
异常处理：
- 如果在循环体内抛出异常（且未 catch 吞掉），Parallel.ForEachAsync会将异常包装进AggregateException抛出。
- 如果你想让某个任务失败不影响其他任务，就在循环体内部try-catch吞掉单次异常。
- 如果收到CancellationToken，必须立即停止并抛出OperationCanceledException。