使用blockingcollection和tasks .net 4 TPL的经典生产者消费者模式

Question

请看下面的伪代码

//Single or multiple Producers produce using below method
    void Produce(object itemToQueue)
    {
        concurrentQueue.enqueue(itemToQueue);
        consumerSignal.set;
    }

    //somewhere else we have started a consumer like this
    //we have only one consumer
    void StartConsumer()
    {
        while (!concurrentQueue.IsEmpty())
        {
            if (concurrentQueue.TrydeQueue(out item))
            {
                //long running processing of item
            }
        }
        consumerSignal.WaitOne();
    }

如何移植我自古以来就使用的这种模式，以使用 taskfactory 创建的任务和 net 4 的新信号功能。换句话说，如果有人要使用 net 4 编写这种模式，它会是什么样子？伪代码很好。如您所见，我已经在使用 .net 4 concurrentQueue。如果可能，我如何使用任务并可能使用一些更新的信号机制。谢谢

感谢 Jon/Dan 解决了我的问题。甜的。 没有像过去那样的手动信号或 while(true) 或 while(itemstoProcess) 类型循环

//Single or multiple Producers produce using below method
 void Produce(object itemToQueue)
 {
     blockingCollection.add(item);
 }

 //somewhere else we have started a consumer like this
 //this supports multiple consumers !
 task(StartConsuming()).Start; 

 void StartConsuming()
 {
     foreach (object item in blockingCollection.GetConsumingEnumerable())
     {
                //long running processing of item
     }
 }

cancellations are handled using cancel tokens

Answer 1

您将使用BlockingCollection<T>。文档中有一个示例。

该类是专门为使这变得微不足道而设计的。

Answer 2

您的第二个代码块看起来更好。但是，启动Task 然后立即等待它是没有意义的。只需调用Take，然后处理直接在消费线程上返回的项目。这就是生产者-消费者模式的意义所在。如果您认为工作项的处理足够密集以保证更多的消费者，那么一定要启动更多的消费者。 BlockingCollection 是安全的多个生产者和多个消费者。

public class YourCode
{
  private BlockingCollection<object> queue = new BlockingCollection<object>();

  public YourCode()
  {
    var thread = new Thread(StartConsuming);
    thread.IsBackground = true;
    thread.Start();
  }

  public void Produce(object item)
  {
    queue.Add(item);
  }

  private void StartConsuming()
  {
    while (true)
    {
      object item = queue.Take();
      // Add your code to process the item here.
      // Do not start another task or thread. 
    }
  }
}

Answer 3

我之前使用过一种模式，它创建了一种“按需”队列消费者（基于来自 ConcurrentQueue 的消费）：

        private void FireAndForget(Action fire)
        {
            _firedEvents.Enqueue(fire);
            lock (_taskLock)
            {
                if (_launcherTask == null)
                {
                    _launcherTask = new Task(LaunchEvents);
                    _launcherTask.ContinueWith(EventsComplete);
                    _launcherTask.Start();
                }
            }
        }

        private void LaunchEvents()
        {
            Action nextEvent;

            while (_firedEvents.TryDequeue(out nextEvent))
            {
                if (_synchronized)
                {
                    var syncEvent = nextEvent;
                    _mediator._syncContext.Send(state => syncEvent(), null);
                }
                else
                {
                    nextEvent();                        
                }

                lock (_taskLock)
                {
                    if (_firedEvents.Count == 0)
                    {
                        _launcherTask = null;
                        break;
                    }
                }
            }
        }

        private void EventsComplete(Task task)
        {
            if (task.IsFaulted && task.Exception != null)
            {
                 // Do something with task Exception here
            }
        }