C# 如何使用Channel - .NET中高性能的生产者消费者队列

Channel 是 .NET 5+ 推荐的异步生产者-消费者通信原语,相比 Queue 和 BlockingCollection 更轻量、支持无锁操作、内置完成与取消感知,具备有界/无界模式以控制背压,Reader/Writer 可分离实现组件解耦,配合 TryRead 批处理与 WriteAsync 等待机制,适用于高并发低延迟场景如实时消息处理与任务管道。

- .net中高性能的生产者消费者队列">

Channel 是 .NET 5+ 中推荐的高性能、异步友好的生产者-消费者通信原语,比 BlockingCollection 更轻量、更灵活,尤其适合高并发、低延迟场景(如实时消息处理、后台任务管道、流式数据处理)。

为什么选 Channel 而不是 Queue 或 BlockingCollection

它不是线程安全的普通队列,而是一个**异步就绪的通道(channel)**,天然支持:

  • 无锁、零分配(在多数配置下)的写入/读取
  • 内置完成(Writer.Complete())和取消感知(可传 CancellationToken
  • 支持“有界”(Bounded)与“无界”(Unbounded)两种模式,可控内存增长
  • Reader 和 Writer 可分离传递,便于解耦组件(比如一个服务只写,另一个只读)

快速上手:创建与基本用法

最简示例(无界 Channel):

var channel = Channel.CreateUnbounded();
var reader = channel.Reader;
var writer = channel.Writer;

// 生产者(异步写入)
_ = Task.Run(async () =>
{
    await writer.WriteAsync("Hello");
    await writer.WriteAsync("World");
    writer.Complete(); // 标记不再写入
});

// 消费者(异步读取)
await foreach (var msg in reader.ReadAllAsync())
{
    Console.WriteLine(msg); // 输出 Hello,然后 World
}

注意:ReadAllAsync() 会自动等待新项、响应完成信号,并在通道关闭后退出循环。

控制背压:使用有界 Channel 防止内存爆炸

当生产快于消费时,无界 Channel 会导致内存无限堆积。改用有界 Channel 可自然施加背压:

// 最多缓存 100 个字符串,超出时 WriteAsync 会 await(阻塞生产者直到有空位)
var channel = Channel.CreateBounded(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.Wait // 默认行为;也可设为 DropWrite / DropOldest
});

常见策略:

  • FullMode = Wait:默认,生产者等待空位(最常用,保证不丢数据)
  • FullMode = DropWrite:新数据直接丢弃(适合监控指标等非关键流)
  • FullMode = DropOldest:挤掉最老的数据腾位置(适合滑动窗口场景)

进阶技巧:手动读取 + 异常处理 + 取消支持

不用 await foreach 时,可用 TryRead(非阻塞)或 ReadAsync(异步阻塞),并配合取消令牌:

var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));

try
{
    while (await reader.WaitToReadAsync(cts.Token))
    {
        while (reader.TryRead(out var item))
        {
            Process(item);
        }
    }
}
catch (OperationCanceledException)
{
    Console.WriteLine("读取被取消");
}
catch (ChannelClosedException)
{
    Console.WriteLine("通道已关闭");
}

关键点:

  • WaitToReadAsync() 等待有新数据或关闭,避免忙等
  • TryRead() 批量消费当前所有可用项(推荐,减少 await 开销)
  • 始终检查 ChannelReader.Completion.IsFaulted 判断是否因异常关闭

基本上就这些。Channel 不复杂但容易忽略细节——重点是选对有界策略、善用 TryRead 批处理、别忘了 Complete() 和取消传播。它不是万能队列,而是为异步流水线设计的“管道”,用对了,吞吐翻倍,延迟归零。