最推荐使用System.Threading.Channels实现生产者-消费者队列。它支持有界和无界通道,提供异步操作与背压机制,适用于多种应用场景,尤其适合现代异步编程模型。

在 .NET 中实现生产者-消费者队列,最推荐的方式是使用 System.Threading.Channels 命名空间中的 Channel 类。它是微软官方为现代异步场景设计的高性能、线程安全的生产者-消费者队列组件,适用于控制台应用、ASP.NET、后台服务等多种场景。
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() 表示不再有新数据,消费者会在所有消息处理完后退出。默认情况下,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 适合异步流处理。
不要在 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 推荐的实现方式。
以上就是.NET如何实现一个生产者-消费者队列的详细内容,更多请关注php中文网其它相关文章!
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移,垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是,许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号