首页 > 编程语言 >C# Parallel.ForEachAsync异步多线程用法

C# Parallel.ForEachAsync异步多线程用法

来源：互联网 2026-07-26 08:10:27

Parallel.ForEachAsync是.NET6引入的异步并发方法，通过MaxDegreeOfParallelism精确控制并发数，避免资源耗尽。支持取消令牌和线程安全集合收集结果，适用于批量网络请求等I/O密集型场景，相比传统方式兼顾效率与稳定性，确保高并发下资源合理利用。

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个异步任务，完成一个再补充一个，保持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
执行方式	串行（一个接一个）	全量并发（同时启动所有）	可控并发（池化执行）
内存/资源占用	极低	极高（容易内存溢出/端口耗尽）	适中（按需启动）
总耗时	最长	最短（若不崩溃）	较短（与并发数成正比）
支持取消	支持	支持	完美支持