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

.NET 中实现生产者-消费者队列，核心目标是让生产者线程安全地向队列添加数据，消费者线程从队列中取出并处理数据，同时避免资源竞争和死锁。.NET 提供了多种方式来高效、安全地实现这一模式。

使用 BlockingCollection

特点：自动处理线程阻塞（当队列为空时消费者等待，满时生产者等待），支持超时、取消、完成添加通知等高级功能。

示例代码：

var queue = new BlockingCollection<string>(new ConcurrentQueue<string>());
// 生产者
Task.Run(() =>
{
    for (int i = 0; i < 5; i++)
    {
        queue.Add("Item " + i);
        Thread.Sleep(100);
    }
    queue.CompleteAdding(); // 通知消费者不再添加
});
// 消费者
Task.Run(() =>
{
    foreach (var item in queue.GetConsumingEnumerable())
    {
        Console.WriteLine("处理: " + item);
    }
});

使用 Channel（.NET Core/.NET 5+）

特点：支持同步/异步读写，可配置有界/无界缓冲，天然集成 async/await。

示例代码：

var channel = Channel.CreateUnbounded<string>();
// 生产者
Task.Run(async () =>
{
    for (int i = 0; i < 5; i++)
    {
        await channel.Writer.WriteAsync("Item " + i);
        await Task.Delay(100);
    }
    channel.Writer.Complete();
});
// 消费者
Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine("处理: " + item);
    }
});

手动实现（不推荐但有助于理解原理）

关键点：必须保护共享队列，使用条件变量（如 Monitor.Wait/Pulse）避免轮询。

简化示例：

private readonly Queue<string> _queue = new();
private readonly object _lock = new();
private volatile bool _completed;
public void Add(string item)
{
    lock (_lock)
    {
        _queue.Enqueue(item);
        Monitor.Pulse(_lock); // 唤醒消费者
    }
}
public bool TryTake(out string item)
{
    lock (_lock)
    {
        while (_queue.Count == 0 && !_completed)
            Monitor.Wait(_lock);
        if (_queue.Count > 0)
        {
            item = _queue.Dequeue();
            return true;
        }
        item = null;
        return false;
    }
}

选择建议