ByteFisher

钓鱼爱好者的编程世界

C# 高级教程 - 07 TPL 与并行编程

发表于 分类于 ， Waline：
C#、高级: Task Parallel Library（TPL）是 .NET 中构建并行和异步应用的核心框架。基础教程中我们接触了 Task 的基本用法和 async/await，但 TPL 的能力远不止于此。

Task Parallel Library（TPL）是 .NET 中构建并行和异步应用的核心框架。基础教程中我们接触了 Task 的基本用法和 async/await，但 TPL 的能力远不止于此。本文将深入探讨 Task 的底层机制、Parallel 类、PLINQ、Dataflow 库、以及如何正确控制并行度与取消。

Task 底层细节

Task 状态机

1
2
3
4
5
6
7
8
9
// Task 的状态转换
Task task = new Task(() => Thread.Sleep(100));
Console.WriteLine(task.Status); // Created

task.Start();
Console.WriteLine(task.Status); // WaitingToRun

task.Wait();
Console.WriteLine(task.Status); // RanToCompletion

TaskCreationOptions 与 TaskScheduler

1
2
3
4
5
6
7
8
9
10
// 长时间运行任务 —— 专用线程（非线程池）
Task longRunning = Task.Factory.StartNew(() =>
{
    Thread.Sleep(TimeSpan.FromHours(1));
}, TaskCreationOptions.LongRunning);

// 亲和性调度 —— 指定在特定调度器上运行
var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();
Task.Factory.StartNew(() => UpdateUI(), CancellationToken.None,
    TaskCreationOptions.None, uiScheduler);

父子任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
Task parent = Task.Factory.StartNew(() =>
{
    Console.WriteLine("父任务开始");

    Task child = Task.Factory.StartNew(() =>
    {
        Thread.Sleep(500);
        Console.WriteLine("子任务完成");
    }, TaskCreationOptions.AttachedToParent);

    Console.WriteLine("父任务结束");
});

parent.Wait(); // 会等待子任务完成
Console.WriteLine("所有任务完成");

Parallel 类

Parallel.For 与 Parallel.ForEach

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 并行 for 循环
Parallel.For(0, 100, i =>
{
    Console.WriteLine($"处理 {i}，线程 {Thread.CurrentThread.ManagedThreadId}");
});

// 并行 foreach
var items = Enumerable.Range(1, 1000);
Parallel.ForEach(items, item =>
{
    Compute(item);
});

// 控制并行度
var options = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount / 2,
    CancellationToken = cancellationToken
};

Parallel.For(0, 100, options, i =>
{
    cancellationToken.ThrowIfCancellationRequested();
    Process(i);
});

Parallel.Invoke

1
2
3
4
5
Parallel.Invoke(
    () => ProcessImage("image1.jpg"),
    () => ProcessImage("image2.jpg"),
    () => GenerateThumbnail("image3.jpg")
);

Break vs Stop

1
2
3
4
5
6
7
8
Parallel.For(0, 100_000, (i, state) =>
{
    if (i > 5000)
    {
        state.Break(); // 停止当前迭代之前的所有后续迭代
        // state.Stop(); // 立即停止所有迭代
    }
});

PLINQ（Parallel LINQ）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
var nums = Enumerable.Range(1, 1_000_000);

// 并行查询
var parallelResult = nums.AsParallel()
    .Where(n => IsPrime(n))
    .Select(n => n * n)
    .ToArray();

// 保留顺序
var orderedResult = nums.AsParallel()
    .AsOrdered()
    .Where(n => n % 2 == 0)
    .ToArray();

// 强制并行（即使查询可能更慢）
var forceParallel = nums.AsParallel()
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .Where(n => ExpensiveCheck(n));

// 控制并行度
var controlled = nums.AsParallel()
    .WithDegreeOfParallelism(4)
    .Select(n => HeavyComputation(n))
    .ToList();

// Merge 选项控制结果缓冲
var merge = nums.AsParallel()
    .WithMergeOptions(ParallelMergeOptions.NotBuffered) // 实时输出
    .Select(n => ExpensiveCompute(n));

foreach (var item in merge)
{
    Console.WriteLine(item); // 每计算完一个就输出一个
}

Dataflow（TPL Dataflow）

System.Threading.Tasks.Dataflow 提供了基于 Actor 模型的数据流管道，非常适合生产者-消费者模式和流水线处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// 需要安装: dotnet add package System.Threading.Tasks.Dataflow

var bufferBlock = new BufferBlock<int>();

var transformBlock = new TransformBlock<int, string>(n =>
{
    Thread.Sleep(100); // 模拟处理
    return $"处理结果: {n * 2}";
}, new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 4,
    BoundedCapacity = 100
});

var actionBlock = new ActionBlock<string>(result =>
{
    Console.WriteLine(result);
});

// 链接块（类似管道）
bufferBlock.LinkTo(transformBlock, new DataflowLinkOptions { PropagateCompletion = true });
transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });

// 生产数据
for (int i = 0; i < 20; i++)
{
    await bufferBlock.SendAsync(i);
}

bufferBlock.Complete();
await actionBlock.Completion;

常用 Dataflow 块

类型 行为
BufferBlock<T> 先进先出队列
TransformBlock<TInput, TOutput> 转换每个元素
ActionBlock<T> 对每个元素执行操作
BatchBlock<T> 将 N 个元素打包为一个批次
JoinBlock<T1, T2> 等待两个来源匹配后合并
BroadcastBlock<T> 广播给所有目标

示例：批处理并行下载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
var downloadBlock = new TransformBlock<string, (string url, byte[] data)>(
    async url => (url, await new HttpClient().GetByteArrayAsync(url)),
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 }
);

var batchBlock = new BatchBlock<(string, byte[])>(10); // 每 10 个一批
var saveBlock = new ActionBlock<(string, byte[])[]>(async batch =>
{
    foreach (var (url, data) in batch)
    {
        var fileName = Path.GetFileName(url);
        await File.WriteAllBytesAsync(fileName, data);
    }
});

downloadBlock.LinkTo(batchBlock, new DataflowLinkOptions { PropagateCompletion = true });
batchBlock.LinkTo(saveBlock, new DataflowLinkOptions { PropagateCompletion = true });

foreach (var url in urls)
{
    await downloadBlock.SendAsync(url);
}
downloadBlock.Complete();
await saveBlock.Completion;

取消令牌（CancellationToken）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
using var cts = new CancellationTokenSource();
CancellationToken token = cts.Token;

var task = Task.Run(() =>
{
    for (int i = 0; i < 100; i++)
    {
        token.ThrowIfCancellationRequested();
        Thread.Sleep(200);
    }
}, token);

// 5 秒后取消
cts.CancelAfter(5000);

try
{
    await task;
}
catch (OperationCanceledException)
{
    Console.WriteLine("任务已取消");
}

链接取消令牌

1
2
3
4
5
6
using var appCts = new CancellationTokenSource();
using var timeoutCts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
    appCts.Token, timeoutCts.Token);

await Task.Run(() => DoWork(linkedCts.Token), linkedCts.Token);

Task 组合模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// WhenAll —— 等待全部完成
var tasks = Enumerable.Range(0, 10).Select(i => ProcessAsync(i));
await Task.WhenAll(tasks);

// WhenAny —— 任意完成即继续
var first = await Task.WhenAny(tasks);
if (first.Status == TaskStatus.RanToCompletion)
{
    await first; // 获取结果（不会阻塞）
}

// 自定义 WhenAll 限流
static async Task<T[]> WhenAllThrottled<T>(
    IEnumerable<Func<CancellationToken, Task<T>>> taskFactories,
    int maxConcurrency, CancellationToken token = default)
{
    var semaphore = new SemaphoreSlim(maxConcurrency);
    var tasks = taskFactories.Select(async factory =>
    {
        await semaphore.WaitAsync(token);
        try
        {
            return await factory(token);
        }
        finally
        {
            semaphore.Release();
        }
    });
    return await Task.WhenAll(tasks);
}

异常处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
Task task = Task.Run(() =>
{
    throw new InvalidOperationException("出错了");
    throw new AccessViolationException("又出错了"); // 不会执行
});

try
{
    await task;
}
catch (AggregateException ae) when (task.IsFaulted)
{
    foreach (var ex in ae.InnerExceptions)
    {
        Console.WriteLine(ex.Message);
    }
}

// 或使用 ContinueWith
task.ContinueWith(t =>
{
    foreach (var ex in t.Exception?.InnerExceptions ?? Enumerable.Empty<Exception>())
    {
        Console.WriteLine($"异常: {ex.Message}");
    }
}, TaskContinuationOptions.OnlyOnFaulted);

总结：TPL 是 .NET 并行编程的现代基石。Parallel 类适合 CPU 密集型数据并行，PLINQ 提供声明式并行查询，Dataflow 库适合复杂的流水线和生产者-消费者模式。理解 Task 的底层机制和各类组合模式，可以帮助我们编写可预测、可取消、可扩展的并行代码。

ByteFisher
分享编程技术 · 记录钓鱼乐趣
扫码关注
▸ 扫码关注 ◂
相关文章
C# 高级教程 - 20 性能优化与诊断2025-05-27C#、高级: 性能优化不是"过早优化"，而是基于数据的科学工程实践。.NET 生态提供了世界一流的诊断工具链——从 BenchmarkDotNet 到 OpenTelemetry，从 dotnet-trace 到 PerfView。本文将系统性地介绍性能分析、诊断工具和优化策略。C# 高级教程 - 19 互操作与源代码生成器2025-05-25C#、高级: 互操作（Interop）让 .NET 代码可以调用原生 C/C++ 库和 COM 组件，这在图像处理、硬件驱动、游戏引擎集成等场景中不可或缺。源代码生成器（Source Generator）则是 Roslyn 编译器的扩展能力，可以在编译时分析代码并生成额外的源文件。本文将深入探讨这两项技术。C# 高级教程 - 18 函数式编程2025-05-23C# 是一门多范式语言，在面向对象基础上不断吸收函数式编程（FP）的特性。从 LINQ 到模式匹配，从 record 到不可变数据，C# 的函数式表达能力持续增强。本文将深入探讨不可变性、纯函数、高阶函数、Either/Option 模式、以及如何在 C# 中实践函数式设计。C# 高级教程 - 17 高级设计模式2025-05-21C#、高级: 基础教程中我们了解过单例、工厂、观察者等经典设计模式。在实际的 C# 项目中，这些模式有更地道的 .NET 实现方式，同时也需要理解模式之间的组合与演进。本文将探讨设计模式在 C# 中的最佳实践以及 .NET 生态中的模式变体。C# 高级教程 - 16 依赖注入2025-05-19C#、高级: 依赖注入（DI）是现代 .NET 应用架构的核心模式。ASP.NET Core 内置了完整的 DI 容器，了解其工作原理、生命周期管理、高级注册方式和常见陷阱，对于构建可测试、可维护的企业级应用至关重要。本文将深入探讨 DI 容器的内部机制和最佳实践。C# 高级教程 - 15 高级异步模式2025-05-17C#、高级: 基础教程中我们学会了 async/await 的基本用法。但在高性能场景下，不当的异步编程会导致性能反而比同步更差。本文将深入探讨 ValueTask、IAsyncEnumerable、Channel、管道模式、以及异步同步上下文等高级话题，帮助你编写真正高效的异步代码。
C#高级教程 系列
第 7/20 篇
C# 高级教程 - 06 多线程基础C# 高级教程 - 08 并发集合与同步
分享：
淡水鱼
淡水鱼
钓鱼爱好者 & Unity 游戏开发者，分享编程技术与钓鱼经验。