如何实现可挂起的BlockingCollection

Question

我正在编写一个WCF服务,它接收来自多个模块(DB,其他服务..)的通知,并将它们添加到阻塞集合中,以便在将相关数据发布到客户端的消费者线程上进行处理.

客户端可以请求存储在服务器上的完整数据,在此操作期间,我不想接受任何新通知.基本上我想暂停阻塞集合(或消费者线程)并在完成客户端请求后恢复通知接收和处理.

实现此行为的好方法是什么？

Answer 1

如果我已正确理解您,您希望阻止使用者线程BlockingCollection在执行某些其他查询操作时使用数据,但在此期间,生产者可以继续将数据推送到集合中.

我这是正确的,那么我认为最好的方法是让一个ManualResetEvent通常发出信号,消费者线程不会被阻止,当你想暂停消费者时,你可以重置事件,这将导致每个消费者阻止等待要发出信号的事件.

更新:这是一个快速的控制台应用程序,演示了我上面描述的内容.这只是一个快速演示,但显示了1个生产者线程和两个消费者线程.消费者的状态之间可以进行切换Running,并Paused通过敲击Space Bar键盘上.

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;

namespace ProducerConsumerDemo
{
  class Program
  {
    static BlockingCollection<int> _queue = new BlockingCollection<int>();
    static ManualResetEvent _pauseConsumers = new ManualResetEvent(true);
    static bool _paused = false;
    static int _itemsEnqueued = 0;
    static int _itemsDequeued = 0;

    static void Main(string[] args)
    {
      Thread producerThread = new Thread(Producer);
      Thread consumerThread1 = new Thread(Consumer);
      Thread consumerThread2 = new Thread(Consumer);
      producerThread.Start();
      consumerThread1.Start();
      consumerThread2.Start();

      while (true)
      {
        WriteAt(0,0,"State: " + (string)(_paused ? "Paused" : "Running"));
        WriteAt(0,1,"Items In Queue: " + _queue.Count);
        WriteAt(0, 2, "Total enqueued: " + _itemsEnqueued);
        WriteAt(0, 3, "Total dequeued: " + _itemsDequeued);

        Thread.Sleep(100);
        if (Console.KeyAvailable)
        {
          if (Console.ReadKey().Key == ConsoleKey.Spacebar)
          {
            if (_paused)
            {
              _paused = false;
              _pauseConsumers.Set();
            }
            else
            {
              _paused = true;
              _pauseConsumers.Reset();
            }
          }
        }
      }
    }

    static void WriteAt(int x, int y, string format, params object[] args)
    {
      Console.SetCursorPosition(x, y);
      Console.Write("                                         ");
      Console.SetCursorPosition(x, y);
      Console.Write(format, args);
    }

    static void Consumer()
    {
      while (true)
      {
        if (_paused)
        {
          // If we are paused, wait for the signal to indicate that
          // we can continue
          _pauseConsumers.WaitOne();
        }

        int value;
        if (_queue.TryTake(out value))
        {
          Interlocked.Increment(ref _itemsDequeued);
          // Do something with the data
        }
        Thread.Sleep(500);
      }
    }

    static void Producer()
    {
      Random rnd = new Random();
      while (true)
      {
        if (_queue.TryAdd(rnd.Next(100)))
        {
          Interlocked.Increment(ref _itemsEnqueued);
        }
        Thread.Sleep(500);
      }
    }
  }
}