C#中使用Parallel.ForEachAsync异步操作多线程的做法实用指南-45看点

平时做技术实践时，很多问题不是概念不会，而是细节没串起来。拿“C#中采用Parallel.ForEachAsync异步操作多线程的做法”来说，它看着像小点，放到项目里常会牵出环境、配置、兼容性和维护成本。下面按实际采用顺序，把思路、关键写法和容易踩坑的地方讲清楚，便于大家直接对照操作。

1. 核心概念

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

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

2. 基本语法与关键参数

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

关键参数说明：

source：数据源（如数组、List、或你用 Chunk() 切分出的块）。
parallelOptions：同时行设置项，极其重要，包含两个核心属性：
- MaxDegreeOfParallelism：最大并发数。比如设为 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. 核心优势对比表

特性	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。

落到代码里，总的来说，C#多线程异步操作适合结合实际项目边做边理解。先抓住核心思路，再逐步补上细节和边界处理，最后效果会更稳定，也更容易复用。