C#中Parallel.ForEachAsync实现异步多线程处理的详细方法与示例

1. 核心概念

Parallel.ForEachAsync 是 .NET 6 中引入的一个实用方法，专为异步任务而生。

一句话概括它的核心价值：让你能够对一组数据执行异步操作，同时精确控制并发数量——也就是同一时间最多跑多少个任务。

它直击了两个经典痛点：

1. 传统的 Parallel.ForEach 不支持 async/await，它天生是为CPU密集型同步任务设计的，遇到异步就傻眼。
2. 直接用 Task.WhenAll(tasks) 虽然能并发，但会一次性把所有任务都扔出去。假如数据有10万条，那瞬间内存和网络端口就会直接爆炸。

2. 基本语法与关键参数

public static Task ForEachAsync(
    this IEnumerable source,
    ParallelOptions parallelOptions,
    Func body
)

参数拆解：

• source：数据源，可以是数组、List，或者用 Chunk() 切分出来的块。
• parallelOptions：并行配置项，这里面有两个核心成员：
  • MaxDegreeOfParallelism：最大并发数。比如设为5，就意味着最多同时跑5个异步任务，跑完一个再补一个，一直保持5个在跑。
  • CancellationToken：取消令牌，用于在用户取消操作时优雅地终止所有任务。
• body：你要执行的异步逻辑，注意它的返回类型是 ValueTask，比 Task 更轻量，性能更好。

3. 经典示例

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

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

using System.Net.Http;
List urls = GetUrls(); // 假设有 100 个 URL
using HttpClient client = new HttpClient();
// 配置并行选项：最大并发 10
var options = new ParallelOptions { MaxDegreeOfParallelism = 10 };
Console.WriteLine("开始并发下载...");
await Parallel.ForEachAsync(urls, options, async (url, ct) =>
{
    try
    {
        // 传入 ct，如果外部取消，当前请求会立即中止
        string content = await client.GetStringAsync(url, ct);
        Console.WriteLine($"成功下载 {url}，长度: {content.Length}");
    }
    catch (OperationCanceledException)
    {
        // 被取消
    }
    catch (Exception ex)
    {
        Console.WriteLine($"下载失败 {url}: {ex.Message}");
    }
});
Console.WriteLine("全部下载完成。");

示例二：收集结果

ForEachAsync 本身没有返回值（返回的是 Task），如果你需要收集处理结果，必须借助线程安全集合，比如 ConcurrentBag 或 ConcurrentQueue。

using System.Collections.Concurrent;
using System.Diagnostics;

public static async Task Example_WithResultCollection()
{
    var inputs = Enumerable.Range(1, 10);
    var results = new ConcurrentBag(); // 线程安全集合

    await Parallel.ForEachAsync(inputs, async i =>
    {
        await Task.Delay(100); // 模拟耗时操作
        int result = i * i;    // 计算逻辑
        results.Add(result);   // 并发添加结果（线程安全）
        
        Console.WriteLine($"{DateTime.Now:mm:ss:fff} - 处理 {i}");
    });

    Console.WriteLine($"结果数量: {results.Count}");
}

示例三：配合取消令牌

在用户关闭窗口或点击“停止”按钮时，优雅地停止处理，而不是粗暴地终止。

public static async Task Example_WithCancellation(CancellationToken cts)
{
    var range = Enumerable.Range(0, 1000);
    
    try
    {
        await Parallel.ForEachAsync(range, 
            new ParallelOptions { CancellationToken = cts, MaxDegreeOfParallelism = 10 }, 
            async (i, ct) =>
            {
                // 如果外部取消，这里会抛出 OperationCanceledException
                await Task.Delay(100, ct); 
                Console.WriteLine($"处理中: {i}");
            });
    }
    catch (OperationCanceledException)
    {
        Console.WriteLine("任务被用户取消。");
    }
}

4. 核心优势对比表

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。