.NET如何实现一个生产者-消费者队列

来源：这里教程网时间：2026-02-21 17:29:08 作者：

在 .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(options);

// 启动3个消费者 for (int i = 0; i { 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(); services.AddSingleton(channel.Reader); services.AddSingleton(channel.Writer); services.AddHostedService(); ```

基本上就这些。优先使用