在 ConcurrentQueue 中尝试出队答案

【问题标题】：Try Dequeue in ConcurrentQueue在 ConcurrentQueue 中尝试出队
【发布时间】：2011-06-28 04:44:12
【问题描述】：

如果队列中没有项目，ConcurrentQueue 中的 TryDequeue 将返回 false。

如果队列为空，我需要我的队列等到将新项目添加到队列中并将新项目从队列中取出，然后该过程将继续进行。

我应该在 C# 4.0 中使用 monitor.enter、wait、pulse 还是任何更好的选项

【问题讨论】：

hmmm - 让我感到惊讶的是，这并没有方便地公开 - 这似乎是一个常见的用例......
我认为使用 4.0 的 BCL 类应该公开这一点，但如果没有其他内容出现，也许看看 at this answer，其中 TryDequeue 将阻塞直到 either 有一个项目 (返回true）或队列为空并显式关闭（返回false）
如果队列为空，我需要我的线程等到新项目被添加到队列中并将新项目出列，然后该过程将继续进行。
@Marc ：添加新项目后，我需要恢复我的线程以退出队列。我怎样才能做到这一点。
上面是这样做的，但看起来“Damien_The_Unbeliever”在这里有答案

标签： c# producer-consumer concurrent-queue

【解决方案1】：

这不就是 BlockingCollection 的用途吗？

据我了解，您可以使用其中之一包装您的 ConcurrentQueue，然后调用 Take。

【讨论】：

【解决方案2】：

您可以使用BlockingCollection。

做这样的事情：

private BlockingCollection<string> rowsQueue;
private void ProcessFiles() {
   this.rowsQueue = new BlockingCollection<string>(new ConcurrentBag<string>(), 1000);
   ReadFiles(new List<string>() { "file1.txt", "file2.txt" });


   while (!this.rowsQueue.IsCompleted || this.rowsQueue.Count > 0)
   {
       string line = this.rowsQueue.Take();

       // Do something
   }
}

private Task ReadFiles(List<string> fileNames)
{
    Task task = new Task(() =>
    {
        Parallel.ForEach(
        fileNames,
        new ParallelOptions
        {
            MaxDegreeOfParallelism = 10
        },
            (fileName) =>
            {
                using (StreamReader sr = File.OpenText(fileName))
                {
                    string line = String.Empty;
                    while ((line = sr.ReadLine()) != null)
                    {
                           this.rowsQueue.Add(line);
                    }
                }
            });

        this.rowsQueue.CompleteAdding();
    });

    task.Start();

    return task;
}

【讨论】：

【解决方案3】：

您可以定期检查队列中的元素数量，当元素数量大于零时，您可以使用例如ManualResetEvent 到使元素出队直到队列为空的线程。

这里是这个的伪代码：

检查线程：

while(true)
{
  int QueueLength = 0;
  lock(Queue)
  {
    queueLength = Queue.Length;
  }

  if (Queue.Length > 0)
  {
    manualResetEvent.Set();
  }
  else
  {
    Thread.Sleep(...);
  }       
}

出列线程：

while(true)
{
  if(manualResetEvent.WaitOne(timeout))
  {
    DequeueUntilQueueEmpty();
  }
}

考虑在 DequeueUntilQueueEmpty 中也使用锁。

【讨论】：

其实我使用 Monitor.enter,wait,pulseall 让线程等待并发出信号。 ManualResetEvent 是更好的选择吗？
其实我使用 Monitor.enter,wait,pulseall 让线程等待并发出信号。 ManualResetEvent 是更好的选择吗？
ManualResetEvent 是轻量级的选项，Monitor 是一个沉重的选项。