異步取消阻止收集

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()時，它會嘗試將所有排隊的項目分配到排隊的回叫。這樣就沒有創建線程，只是坐在那裏睡覺，等待被髮信號。

該鎖定有點醜陋，但您可以在不鎖定的情況下排隊並在多個線程上訂閱。我找不到一種方法來執行相當於的操作，只有在其他隊列上沒有使用該鎖的情況下才可以退出。

注意：我只用最少的幾個線程測試了它。