异步取消阻止收集

Question

我正在使用BlockingCollection来实现生产者/消费者模式。我有一个异步循环，用要处理的数据填充集合，然后在稍后时间由客户端访问。数据包到达稀疏，我希望轮询在不使用阻塞呼叫的情况下完成。

实质上，我正在寻找类似BeginTake和EndTake的东西，它不存在于阻塞集合中，以便我可以在回调中使用内部线程池。无论如何，它不一定是BlockingCollection。任何我需要的东西都会很棒。

这就是我现在得到的。 _bufferedPackets是BlockingCollection<byte[]>：

public byte[] Read(int timeout) 
{ 
    byte[] result; 
    if (_bufferedPackets.IsCompleted) 
    { 
     throw new Exception("Out of packets"); 
    } 
    _bufferedPackets.TryTake(out result, timeout);  
    return result; 
} 

我想这是这样的，在伪代码：

public void Read(int timeout) 
{ 
    _bufferedPackets.BeginTake(result => 
     { 
      var bytes = _bufferedPackets.EndTake(result); 
      // Process the bytes, or the resuting timeout 
     }, timeout, _bufferedPackets); 
} 

什么是我对这个选择？我不希望任何线程处于等待状态，因为有很多其他IO东西要处理，并且我会很快耗尽线程。

更新：我已经重写有问题的代码，以不同的方式使用异步过程，实质上交换基础上的回调，如果有超时限制内等待请求。这工作正常，但如果有一种方法可以做到这一点，而不依赖于定时器并交换周围可能导致竞争条件且很难写入（并理解）的lambda的方法，它仍然会很棒。我已经用自己的异步队列实现解决了这个问题，但如果有更标准和经过充分测试的选项，它仍然非常棒。

Answer 1

所以看起来并不是一个内置的选择，我出去努力做我想做的实验。事实证明，为了使这项工作与旧的异步模式的其他用户大致相同，需要做大量的工作。

public class AsyncQueue<T> 
{ 
    private readonly ConcurrentQueue<T> queue; 
    private readonly ConcurrentQueue<DequeueAsyncResult> dequeueQueue; 

    private class DequeueAsyncResult : IAsyncResult 
    { 
     public bool IsCompleted { get; set; } 
     public WaitHandle AsyncWaitHandle { get; set; } 
     public object AsyncState { get; set; } 
     public bool CompletedSynchronously { get; set; } 
     public T Result { get; set; } 

     public AsyncCallback Callback { get; set; } 
    } 

    public AsyncQueue() 
    { 
     dequeueQueue = new ConcurrentQueue<DequeueAsyncResult>(); 
     queue = new ConcurrentQueue<T>(); 
    } 

    public void Enqueue(T item) 
    { 
     DequeueAsyncResult asyncResult; 
     while (dequeueQueue.TryDequeue(out asyncResult)) 
     { 
      if (!asyncResult.IsCompleted) 
      { 
       asyncResult.IsCompleted = true; 
       asyncResult.Result = item; 

       ThreadPool.QueueUserWorkItem(state => 
       { 
        if (asyncResult.Callback != null) 
        { 
         asyncResult.Callback(asyncResult); 
        } 
        else 
        { 
         ((EventWaitHandle) asyncResult.AsyncWaitHandle).Set(); 
        } 
       }); 
       return; 
      } 
     } 
     queue.Enqueue(item); 
    } 

    public IAsyncResult BeginDequeue(int timeout, AsyncCallback callback, object state) 
    { 
     T result; 
     if (queue.TryDequeue(out result)) 
     { 
      var dequeueAsyncResult = new DequeueAsyncResult 
      { 
       IsCompleted = true, 
       AsyncWaitHandle = new EventWaitHandle(true, EventResetMode.ManualReset), 
       AsyncState = state, 
       CompletedSynchronously = true, 
       Result = result 
      }; 
      if (null != callback) 
      { 
       callback(dequeueAsyncResult); 
      } 
      return dequeueAsyncResult; 
     } 

     var pendingResult = new DequeueAsyncResult 
     { 
      AsyncState = state, 
      IsCompleted = false, 
      AsyncWaitHandle = new EventWaitHandle(false, EventResetMode.ManualReset), 
      CompletedSynchronously = false, 
      Callback = callback 
     }; 
     dequeueQueue.Enqueue(pendingResult); 
     Timer t = null; 
     t = new Timer(_ => 
     { 
      if (!pendingResult.IsCompleted) 
      { 
       pendingResult.IsCompleted = true; 
       if (null != callback) 
       { 
        callback(pendingResult); 
       } 
       else 
       { 
        ((EventWaitHandle)pendingResult.AsyncWaitHandle).Set(); 
       } 
      } 
      t.Dispose(); 
     }, new object(), timeout, Timeout.Infinite); 

     return pendingResult; 
    } 

    public T EndDequeue(IAsyncResult result) 
    { 
     var dequeueResult = (DequeueAsyncResult) result; 
     return dequeueResult.Result; 
    } 
} 

我不是太肯定的IsComplete属性的同步，而我不是如何dequeueQueue只拿到后续Enqueue调用清理太热。我不确定什么时候是正确的时间来等待手柄，但这是迄今为止我所得到的最佳解决方案。

请不要以任何方式考虑此生产质量代码。我只是想展示一下我如何保持所有线程不用等待锁定而旋转的一般要点。我确信这充满了各种各样的边缘情况和错误，但它满足了要求，我想回馈给那些遇到问题的人。

Answer 2

我可能会误解你的情况，但是你不能使用非阻塞集合吗？

我创造了这个例子来说明：

using System; 
using System.Collections.Concurrent; 
using System.Threading; 
using System.Threading.Tasks; 

namespace AsyncTakeFromBlockingCollection 
{ 
    class Program 
    { 
     static void Main(string[] args) 
     { 
      var queue = new ConcurrentQueue<string>(); 

      var producer1 = Task.Factory.StartNew(() => 
      { 
       for (int i = 0; i < 10; i += 1) 
       { 
        queue.Enqueue("======="); 
        Thread.Sleep(10); 
       } 
      }); 

      var producer2 = Task.Factory.StartNew(() => 
      { 
       for (int i = 0; i < 10; i += 1) 
       { 
        queue.Enqueue("*******"); 
        Thread.Sleep(3); 
       } 
      }); 

      CreateConsumerTask("One ", 3, queue); 
      CreateConsumerTask("Two ", 4, queue); 
      CreateConsumerTask("Three", 7, queue); 

      producer1.Wait(); 
      producer2.Wait(); 
      Console.WriteLine(" Producers Finished"); 
      Console.ReadLine(); 
     } 

     static void CreateConsumerTask(string taskName, int sleepTime, ConcurrentQueue<string> queue) 
     { 
      Task.Factory.StartNew(() => 
      { 
       while (true) 
       { 
        string result; 
        if (queue.TryDequeue(out result)) 
        { 
         Console.WriteLine(" {0} consumed {1}", taskName, result); 
        } 
        Thread.Sleep(sleepTime); 
       } 
      }); 
     } 
    } 
} 

这里是程序的输出

我相信BlockingCollection旨在包裹并发收集和提供一种机制，允许多个消费者阻止;等待生产者。这种用法似乎与您的要求相悖。

我发现这article about the BlockingCollection class是有帮助的。

Answer 3

我很确定BlockingCollection<T>不能做到这一点，你不得不推出自己的。我想出了这个：

class NotifyingCollection<T> 
{ 
    private ConcurrentQueue<Action<T>> _subscribers = new ConcurrentQueue<Action<T>>(); 
    private ConcurrentQueue<T> _overflow = new ConcurrentQueue<T>(); 

    private object _lock = new object(); 

    public void Add(T item) 
    { 
     _overflow.Enqueue(item); 
     Dispatch(); 
    } 

    private void Dispatch() 
    { 
     // this lock is needed since we need to atomically dequeue from both queues... 
     lock (_lock) 
     { 
      while (_overflow.Count > 0 && _subscribers.Count > 0) 
      { 
       Action<T> callback; 
       T item; 

       var r1 = _overflow.TryDequeue(out item); 
       var r2 = _subscribers.TryDequeue(out callback); 

       Debug.Assert(r1 && r2); 
       callback(item); 
       // or, optionally so that the caller thread's doesn't take too long ... 
       Task.Factory.StartNew(() => callback(item)); 
       // but you'll have to consider how exceptions will be handled. 
      } 
     } 
    } 

    public void TakeAsync(Action<T> callback) 
    { 
     _subscribers.Enqueue(callback); 
     Dispatch(); 
    } 
} 

我用它调用TakeAsync()或Add()充当回调线程的线程。当您拨打Add()或TakeAsync()时，它会尝试将所有排队的项目分配到排队的回叫。这样就没有创建线程，只是坐在那里睡觉，等待被发信号。

该锁定有点丑陋，但您可以在不锁定的情况下排队并在多个线程上订阅。我找不到一种方法来执行相当于的操作，只有在其他队列上没有使用该锁的情况下才可以退出。

注意：我只用最少的几个线程测试了它。