.NET如何实现一个生产者-消费者队列_生产者消费者队列实现方案-C#.Net教程-PHP中文网

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

小老鼠

发布： 2025-11-03 19:07:02

原创

272人浏览过

推荐使用 BlockingCollection<T> 或 Channel<T> 实现生产者-消费者队列。BlockingCollection<T> 基于 ConcurrentQueue<T>，提供线程安全与阻塞操作，支持完成通知、超时和取消；示例中生产者添加项并调用 CompleteAdding()，消费者通过 GetConsumingEnumerable() 枚举处理。Channel<T>（.NET Core+）更适用于异步场景，支持有界/无界缓冲，集成 async/await，生产者 WriteAsync 写入，消费者 ReadAllAsync 读取。手动实现使用 lock、Monitor.Wait/Pulse 可控但易错，仅用于学习。选择建议：优先 Channel<T>（异步、现代应用），其次 BlockingCollection<T>（同步、.NET Framework），避免手动实现。

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

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

使用 BlockingCollection<T>

这是最推荐的方式，封装了线程安全的集合与阻塞操作，底层基于 ConcurrentQueue<T> 或其它 IProducerConsumerCollection<T> 实现。

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

示例代码：

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<T>（.NET Core/.NET 5+）

Channel 是现代异步流处理的首选，特别适合高并发、异步场景，由 System.Threading.Channels 提供。

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

歌者PPT

歌者PPT，AI 写 PPT 永久免费

197

查看详情

示例代码：

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);
    }
});

登录后复制

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

使用 lock + Queue + Monitor 或 SemaphoreSlim 控制同步，适用于学习或特殊控制需求。

关键点：必须保护共享队列，使用条件变量（如 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;
    }
}

登录后复制