多个生产者单个消费者锁定方案 [英] Multiple producers single consumer locking schema

问题描述

我有多个生产线程，并且一次消费.在C#中，我为此使用ConcurrentQueue.

I have multiple producing threads, and single consuming. In C#, I'm using ConcurrentQueue for that.

当队列为空时，如何正确使使用者线程进入睡眠状态?

How can I properly put the consumer thread to sleep, when the queue is empty?

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

我尝试了这个，但是有机会

I tried this, but there is some chance, that

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

1. thread B fails to read from _eventQueue
2. thread A writes into _eventQueue
3. thread A sets the signal
4. thread B resets the signal
5. thread B waits indefinitely

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

How to overcome this? Previously, I used lock() and Monitor.Wait(). AutoResetEvent may help (it resets on successful Wait), but it does not support CancellationToken.

推荐答案

您可以使用

You can use the BlockingCollection class to support the case of multiple producers and single consumer.

创建类型为BlockingCollection的对象，如下所示:

Create an object of type BlockingCollection like this:

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

生产者可以简单地调用Add方法，以将一个项目添加到集合中，如下所示:

Producers can simple call the Add method to add an item to the collection like this:

collection.Add("value");

并且消费者可以使用GetConsumingEnumerable方法来获取IEnumerable<T>，该IEnumerable<T>从集合中获取物品.当没有更多项目时，此类枚举将阻止(等待更多项目).此方法还支持取消.

And the consumer can use the GetConsumingEnumerable method to get an IEnumerable<T> that takes items from the collection. Such enumerable will block (wait for more items) when there are no more items. This method also supports cancellation.

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

如果您致电 CompleteAdding 方法，则一旦没有更多项，消耗枚举将结束.

If you call the CompleteAdding method, then the consuming enumerable will finish once there are no more items.

此类完全是线程安全的.

This class is completely thread-safe.

这篇关于多个生产者单个消费者锁定方案的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！