从 C# 中的 ConcurrentQueue 出列对象

Question

嘿，我正在尝试在 C# 中为异步服务器实现 ConcurrentQueue。收到完整消息后，项目将立即排队。为了使消息出列，我正在制作少量线程来完成出列和服务请求的工作。这是不够的，因为每个线程都使用一个 while 循环，这会消耗相当多的处理器时间，原因很明显。

有谁知道在需要时使消息出列但不消耗太多处理时间的方法。

{
    ...

    for (int i = 0; i < 3; i++)
    {
        Thread t = new Thread(new ThreadStart(startParsingMessages));
        t.Start();
    }

    ...
}

private void startParsingMessages()
{
    QueueContainer dequeued = null;
    Console.WriteLine("Trying");
    while (true)
    {
        if (queue.TryDequeue(out dequeued))
        {
            Console.WriteLine("processing queue");
            ProcessMessage(dequeued.socket, dequeued.message);
        }
    }
}

Answer 1

您是否尝试过将其包装在BlockingCollection<T> 中，而不是直接使用ConcurrentQueue<T>？然后你可以使用TryTake(out T, TimeSpan) 等。我相信这是预期的用途：并发集合本身通常会在那里，这样你就可以选择阻塞集合的工作方式。

当然，这不一定是这些集合的唯一用途，但特别是对于 ConcurrentQueue<T>，生产者/消费者队列场景是最常见的场景 - 此时 @987654326 @ 是让做正确事情变得容易的方法。

Answer 2

并发编程中有一种众所周知的模式，称为并行生产者/消费者模式。本文可能有助于解决您的一些问题。 http://blogs.msdn.com/b/csharpfaq/archive/2010/08/12/blocking-collection-and-the-producer-consumer-problem.aspx

Answer 3

您可以使用静态锁定对象让线程等待，然后在准备好处理某些内容时对其进行脉冲。

static readonly object queueLock = new object();

// Threads that enqueue an object
void QueueMessage() {
  lock (queueLock) {
    queue.Enqueue(obj);
    Monitor.Pulse(queueLock);
  }
}
// Thread dequeuer
private void startParsingMessages() {
  QueueContainer dequeued = null;
  Console.WriteLine("Trying");
  while (true) {
    lock(queueLock) {
      if (!queue.TryDequeue(out dequeued)) {
        Console.WriteLine("No object to dequeue, waiting...");
        // Threads will wait here and only one will be released when .Pulse()d
        Monitor.Wait(queueLock);
        dequeued = queue.Dequeue();
      }
      if (dequeued != null) {
        Console.WriteLine("processing queue");
        ProcessMessage(dequeued.socket, dequeued.message);
      }
    }
  }
}

请记住，随着您拥有的分支越多，这可能会变得相当复杂，但其要点是，您锁定一个公共对象并调用Monitor.Wait 以等待一个对象成为Pulsed。当这种情况发生时，只有一个正在等待它的线程将被释放。如果您将大量对象排入队列并希望它们全部离开，您可以调用Monitor.PulseAll，这将释放所有等待对象的线程。

Answer 4

BlockingCollection 解决方案很棒。否则，在大多数情况下，简单地睡觉可能就足够了。它在响应时间上引入了一个小的延迟，并且确实消耗了少量的处理时间。优点是更强大的系统，更容易理解并且不太可能失败：

private void startParsingMessages()
{
    QueueContainer dequeued = null;
    Console.WriteLine("Trying");
    while (true)
    {
        //As long as there is data, process it as fast as possible.
        while(queue.TryDequeue(out dequeued))
        {
            Console.WriteLine("processing queue");
            ProcessMessage(dequeued.socket, dequeued.message);
        }
        //No more data, wait 100 ms before checking again
        Thread.Sleep(100);
    }
}