C# (.NET) 任务队列：Channels 和 TPL Dataflow

C# (.NET) 的任务队列：System.Threading.Channels 和 TPL Dataflow

在开发 C# (.NET) 多线程应用程序时，经常需要组织生产者（producers）和消费者（consumers）之间的异步数据传输。经典的 生产者-消费者 模式可以通过任务队列来解决。在 .NET 生态系统中，有两个强大的工具可以实现这一点：System.Threading.Channels（自 .NET Core 3.0 起引入）和 TPL Dataflow 库（自 .NET 4.5 起成为 .NET 的一部分）。

在本文中，我们将分析这两种方法，它们的特点，并提供实用的代码示例。

1. System.Threading.Channels — 轻量且快速的队列

System.Threading.Channels 命名空间提供了线程安全、异步的队列，针对高吞吐量场景进行了优化。主要类型：

Channel<T> — 通道的抽象类。
Channel.CreateUnbounded<T>() — 无界队列（存在内存溢出风险）。
Channel.CreateBounded<T>(int capacity) — 有界队列，具有背压控制。
ChannelWriter<T> — 用于写入的接口。
ChannelReader<T> — 用于读取的接口。

示例：使用 Channel 的生产者-消费者模式

创建一个简单的示例，其中一个生产者生成数字，一个消费者处理它们。

using System.Threading.Channels;using System.Threading.Tasks;

var channel = Channel.CreateUnbounded<int>();

// Producervar producer = Task.Run(async () =>{    for (int i = 0; i < 10; i++)    {        await channel.Writer.WriteAsync(i);        Console.WriteLine($"已生产: {i}");        await Task.Delay(100);    }    channel.Writer.Complete(); // 发送完成信号});

// Consumervar consumer = Task.Run(async () =>{    await foreach (var item in channel.Reader.ReadAllAsync())    {        Console.WriteLine($"已处理: {item * 2}");        await Task.Delay(200);    }});

await Task.WhenAll(producer, consumer);Console.WriteLine("完成!");

Channels 的优势：

最小的开销（无需分配额外对象）。
内置支持 async/await 和 IAsyncEnumerable。
非常适合高负载系统（例如，处理网络数据包）。

2. TPL Dataflow — 强大的处理管道

System.Threading.Tasks.Dataflow 库提供了更复杂的块，用于构建管道（pipeline）和数据处理网络。主要块：

BufferBlock<T> — 简单的队列。
ActionBlock<T> — 对每个元素执行操作的块。
TransformBlock<TInput, TOutput> — 转换每个元素。
BatchBlock<T> — 将元素分组为批次。

示例：使用 ActionBlock 的管道

实现相同的任务：生产者发送数字，消费者处理。

using System.Threading.Tasks.Dataflow;

var actionBlock = new ActionBlock<int>(async item =>{    await Task.Delay(200);    Console.WriteLine($"已处理: {item * 2}");}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });

// Producerfor (int i = 0; i < 10; i++){    await actionBlock.SendAsync(i);    Console.WriteLine($"已生产: {i}");    await Task.Delay(100);}

actionBlock.Complete();await actionBlock.Completion;Console.WriteLine("完成!");

TPL Dataflow 的优势：