最推荐使用 System.Threading.channel s 实现生产者 - 消费者队列。它支持有界和无界通道，提供 异步 操作与背压机制，适用于多种应用场景，尤其适合现代异步编程模型。
在 .net 中实现生产者 - 消费者队列，最推荐的方式是使用 System.Threading.Channels 命名空间 中的 Channel 类。它是 微软 官方为现代异步场景设计的高性能、线程 安全的生产者 - 消费者队列组件，适用于控制台应用、ASP.NET、后台服务等多种场景。
1. 使用 System.Threading.Channels 实现
Channel 提供了两种模式：有界（Bounded）和无界（Unbounded）。它支持同步和异步操作，天然适配 async/await。
安装包（.NET 6 以下可能需要）：
Install-Package System.Threading.Channels示例：异步生产者 - 消费者模型
创建一个有界通道，启动一个生产者任务和一个消费者任务：
using System; using System.Threading.Channels; using System.Threading.Tasks; var channel = Channel.CreateBounded<string>(10); // 最多存放 10 个消息 // 生产者 _ = Task.Run(async () => {for (int i = 1; i <= 5; i++) {await channel.Writer.WriteAsync($" 消息 {i}"); console.WriteLine($" 生产： 消息 {i}"); await Task.Delay(100); // 模拟耗时 } channel.Writer.Complete(); // 关闭写入端}); // 消费者 _ = Task.Run(async () => {await foreach (var msg in channel.Reader.ReadAllAsync()) {Console.WriteLine($" 消费： {msg}"); await Task.Delay(150); // 模拟处理时间 } Console.WriteLine(" 消费完成。"); }); // 等待一段时间让任务执行 await Task.Delay(2000);说明：
-
WriteAsync是线程安全的，当通道满时会自动等待。 -
ReadAllAsync返回 IAsyncEnumerable，自动监听新消息。 - 调用
Complete()表示不再有新数据，消费者会在所有消息处理完后退出。
2. 多生产者或多消费者场景
默认情况下，Channel 支持多个生产者，但只有一个消费者能安全读取。若需多消费者，应使用 UnboundedChannelOptions 或确保 并发 控制。
允许多个消费者竞争消费：
“`csharp var options = new BoundedChannelOptions(10) {FullMode = BoundedChannelFullMode.Wait, SingleReader = false, // 允许多个消费者 SingleWriter = false // 允许多个生产者};
var channel = Channel.CreateBounded
// 启动 3 个消费者 for (int i = 0; i < 3; i++) {var consumerId = i; _ = Task.Run(async () => {await foreach (var item in channel.Reader.ReadAllAsync()) {Console.WriteLine($” 消费者 {consumerId} 处理： {item}”); await Task.Delay(50); } }); }
<H3>3. 替代方案：BlockingCollection（传统方式）</H3> <p> 在较老项目或不支持 Channel 的框架中，可以使用 <code>BlockingCollection<T></code> 配合 <code>ConcurrentQueue<T></code>。</p> ```csharp using System.Collections.Concurrent; using System.Threading.Tasks; var collection = new BlockingCollection<string>(new ConcurrentQueue<string>()); // 生产者 _ = Task.Run(() => { for (int i = 1; i <= 5; i++) {collection.Add($" 消息 {i}"); Console.WriteLine($" 生产： {i}"); Task.Delay(100).Wait();} collection.CompleteAdding(); // 停止添加}); // 消费者 _ = Task.Run(() => { foreach (var msg in collection.GetConsumingEnumerable()) {Console.WriteLine($" 消费： {msg}"); Task.Delay(150).Wait();} });注意：BlockingCollection 是同步阻塞的，不如 Channel 适合异步流处理。
4. 在 ASP.NET 中的应用建议
不要在 Web 请求中直接启动后台任务。建议结合 IHostedService 或使用 BackgroundService 来运行消费者。
public class MessageConsumerService : BackgroundService {private readonly ChannelReader<string> _reader; public MessageConsumerService(ChannelReader<string> reader) {_reader = reader;} protected override async Task ExecuteAsync(CancellationToken stoppingToken) {await foreach (var msg in _reader.ReadAllAsync(stoppingToken)) {Console.WriteLine($" 后台处理： {msg}"); } } }注册服务：
“`csharp var channel = Channel.CreateUnbounded
基本上就这些。优先使用 System.Threading.Channels，它简洁、高效、支持背压（backpressure），是现代 .NET 推荐的实现方式。
