c# - 多个生产者单个消费者锁定模式

Question

我有多个生产线程和单个消费线程。在 C# 中，我使用 ConcurrentQueue 来实现此目的。

当队列为空时，如何正确让消费者线程进入休眠状态？

ManualResetEventSlim signal;
void WorkerThread(CancellationToken token)
{
    while(!token.IsCancellationRequested)
    {
        object work;
        if (!_eventQueue.TryDequeue(out work))
        {
            signal.Reset();
            signal.Wait(token);
            continue;
        }
        ...
    }
}
...
void Produce(object o)
{
    _eventQueue.Enqueue(o);
    signal.Set();
}

我尝试过这个，但有一些机会，

1. 线程 B 无法从 _eventQueue 读取数据
2. 线程 A 写入 _eventQueue
3. 线程A设置信号
4. 线程 B 重置信号
5. 线程 B 无限期等待

如何克服这个问题？以前，我使用了 lock() 和 Monitor.Wait()。 AutoResetEvent 可能会有所帮助(它会在成功 Wait 时重置)，但它不支持 CancellationToken。

Answer 1

您可以使用BlockingCollection类来支持多个生产者和单个消费者的情况。

创建 BlockingCollection 类型的对象像这样:

BlockingCollection<object> collection = new BlockingCollection<object>(); //You can have a specific type instead of object if you want

生产者可以简单地调用Add将项目添加到集合的方法如下:

collection.Add("value");

消费者可以使用GetConsumingEnumerable获取 IEnumerable<T> 的方法从集合中获取项目。当没有更多项目时，此类枚举将阻塞(等待更多项目)。这个方法也supports cancellation .

foreach (var item in collection.GetConsumingEnumerable())
{
    //Consume item
}

如果您调用 CompleteAdding 方法，那么一旦没有更多的项目，消耗的枚举就会完成。

此类是完全线程安全的。