Очередь <T>.Dequeue возвращает ноль

Question

У меня есть сценарий, где

• несколько потоков помещают данные в очередь

• только ОДИН поток обрабатывает данные, используя код ниже

код -

  while ( Continue )
  {
        while ( queue.Count > 0 )
        {
             MyObj o = queue.Dequeue();
             someProcess(o);
        }
        myAutoResetEvent.WaitOne();
  }

Но иногда queue.Dequeue () возвращает ноль в сценарии выше Что дает?

Answer 1

Вы говорите:

несколько потоков помещают данные в очередь

Метод Queue<T>.Enqueue не является потокобезопасным. Это означает, что работа выполняется в методе Enqueue, который необходимо синхронизировать, если его вызывают несколько потоков. Простым примером будет обновление свойства Count. Можно с уверенностью сказать, что где-то в методе Enqueue есть строка, которая выглядит примерно так:

++count;

Но, как мы все знаем, это не атомарная операция. Это действительно больше похоже на это (с точки зрения того, что на самом деле происходит ):

int newCount = count + 1;
count = newCount;

Так, скажем, count в настоящее время 5, а поток 1 проходит int newCount = count + 1 ... тогда поток 1 думает: «Хорошо, счетчик теперь 5, поэтому я сделаю его 6». Но самая следующая выполняемая операция - это когда поток 2 добирается до int newCount = count + 1 и думает так же, как поток 1 («теперь счетчик равен 6»). Таким образом, два элемента были добавлены в очередь, но их количество изменилось с 5 до 6.

Это просто очень простой пример того, как не потокобезопасный метод, такой как Queue<T>.Enqueue, может быть испорчен, когда доступ не синхронизирован. Это конкретно не объясняет, что происходит в вашем вопросе; я хочу просто указать, что то, что вы делаете, не является поточно-ориентированным и приведет к неожиданному поведению .

Answer 2

Вам нужно прочитать этот пост .

Кроме того, вот очень минимальный каркас "канала" для связи между потоками:

public class Channel<T>
{
    private readonly Queue<T> _queue = new Queue<T>();

    public void Enqueue(T item)
    {
        lock (_queue)
        {
            _queue.Enqueue(item);
            if (_queue.Count == 1)
                Monitor.PulseAll(_queue);
        }
    }

    public T Dequeue()
    {
        lock (_queue)
        {
            while (_queue.Count == 0)
                Monitor.Wait(_queue);

            return _queue.Dequeue();
        }
    }
}

Answer 3

Вам необходимо синхронизировать доступ к очереди. Поместите операторы lock вокруг всех участков кода, которые обращаются к очереди (как чтение, так и запись). Если вы обращаетесь к очереди одновременно из нескольких потоков, внутренние структуры могут быть повреждены, и может произойти почти все.

Answer 4

Хаффа верна, чтение и запись в очередь из нескольких потоков вызовет проблемы, поскольку очередь не является поточно-ориентированной.

Если вы используете .NET 4, используйте класс ConcurrentQueue , который является поточно-ориентированным. Если вы не используете .NET 3 или более раннюю версию, вы можете либо сделать свою собственную блокировку, как указал Гуффа, либо использовать стороннюю библиотеку.

Answer 5

Убедитесь, что ничто не помещает null значения в очередь. null s допускаются как значения в очереди. Кроме того, согласно этого документа , только статические члены Queue<T> являются поточно-ориентированными, поэтому остерегайтесь чтения и записи между потоками.

Answer 6

Если вы используете не универсальную очередь (не то, что я советую ее использовать), вы можете использовать метод Queue.Synchronized для получения потоково-безопасной оболочки:

Queue queue = Queue.Synchronized(new Queue());

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

Answer 7

Почему бы вам не решить проблему таким образом?

while (IsRunning)
{
    do
    {
        MyObj myObj = queue.Dequeue();
        if (myObj != null)
        {
            DoSomethingWith(myObj);
        }
    } while (myObj != null);

    myAutoResetEvent.WaitOne();
}

Обновление

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