Параллельные коллекции едят слишком много процессора без Thread.Sleep

Каким будет правильное использование: BlockingCollection или ConcurrentQueue, чтобы вы могли свободно деактивировать элементы, не выделяя половину или более вашего процессора с помощью потока?

Я запускал некоторые тесты, используя 2 потока, и если у меня не было Thread.Sleep, по крайней мере, 50 ~ 100 мс, он всегда ударил бы как минимум 50% моего процессора.

Вот вымышленный пример:

private void _DequeueItem()
{
    object o = null;
    while(socket.Connected)
    {
        while (!listOfQueueItems.IsEmpty)
        {
            if (listOfQueueItems.TryDequeue(out o))
            {
                // use the data
            }
        }
    }
}

В приведенном выше примере мне нужно будет установить thread.sleep, поэтому процессор не взорвется.

Примечание. Я также пробовал его без проверки IsEmpty, результат был таким же.

Ответ 1

Это не из-за BlockingCollection или ConcurrentQueue, а цикла while:

while(socket.Connected)
{
    while (!listOfQueueItems.IsEmpty)
    { /*code*/ }
}

Конечно, это займет процессор; из-за того, что очередь пуста, тогда цикл while выглядит так:

while (true) ;

который, в свою очередь, будет потреблять ресурсы процессора.

Это не очень хороший способ использовать ConcurrentQueue, вы должны использовать AutoResetEvent с ним, поэтому всякий раз, когда добавляется элемент, вы будете уведомлены. Пример:

private ConcurrentQueue<Data> _queue = new ConcurrentQueue<Data>();
private AutoResetEvent _queueNotifier = new AutoResetEvent(false);

//at the producer:
_queue.Enqueue(new Data());
_queueNotifier.Set();

//at the consumer:
while (true)//or some condition
{
    _queueNotifier.WaitOne();//here we will block until receive signal notification.
    Data data;
    if (_queue.TryDequeue(out data))
    {
        //handle the data
    }
}

Для хорошего использования BlockingCollection вы должны использовать GetConsumingEnumerable() для ожидания добавления элементов, Like:

//declare the buffer
private BlockingCollection<Data> _buffer = new BlockingCollection<Data>(new ConcurrentQueue<Data>());

//at the producer method:
_messageBuffer.Add(new Data());

//at the consumer
foreach (Data data in _buffer.GetConsumingEnumerable())//it will block here automatically waiting from new items to be added and it will not take cpu down 
{
    //handle the data here.
}

Ответ 2

В этом случае вы действительно хотите использовать класс BlockingCollection. Он предназначен для блокировки, пока в очереди не появится элемент. Коллекция такого рода часто упоминается как блокирующая очередь. Эта конкретная реализация безопасна для нескольких производителей и нескольких потребителей. Это то, что удивительно сложно получить, если вы попытались реализовать его самостоятельно. Вот как выглядел бы ваш код, если бы вы использовали BlockingCollection.

private void _DequeueItem()
{
    while(socket.Connected)
    {
        object o = listOfQueueItems.Take();
        // use the data
    }
}

Метод Take блокируется автоматически, если очередь пуста. Он блокируется способом, который помещает поток в состояние SleepWaitJoin, чтобы он не потреблял ресурсы ЦП. Оптимальная вещь о BlockingCollection заключается в том, что она также использует стратегии с низким уровнем блокировки для повышения производительности. Это означает, что Take будет проверять, есть ли элемент в очереди, а если нет, то он ненадолго выполнит ожидание вращения, чтобы предотвратить контекстный переключатель потока. Если очередь все еще пуста, то она будет помещать поток в режим сна. Это означает, что BlockingCollection будет иметь некоторые преимущества производительности, которые ConcurrentQueue обеспечивает в отношении одновременного выполнения.

Ответ 3

Вы можете вызывать Thread.Sleep() только тогда, когда очередь пуста:

private void DequeueItem()
{
    object o = null;

    while(socket.Connected)
    {
        if (listOfQueueItems.IsEmpty)
        {
            Thread.Sleep(50);
        }
        else if (listOfQueueItems.TryDequeue(out o))
        {
            // use the data
        }
    }
}

В противном случае вам следует рассмотреть возможность использования событий.