从阻塞集合中异步获取

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 内容要处理，而且我很快就会用完线程。

更新：我已经重写了相关代码，以不同方式使用异步进程，本质上是根据超时限制内是否存在等待请求来交换回调。这工作得很好，但如果有一种方法可以做到这一点，而无需求助于计时器和交换可能导致竞争条件并且难以编写（和理解）的 lambdas，它仍然会很棒。我也通过自己的异步队列实现解决了这个问题，但如果有一个更标准且经过良好测试的选项，它仍然会很棒。

Answer 1

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

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

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 2

所以看起来没有内置选项，我出去尝试尽我所能做我想要的实验。事实证明，要使这项工作与旧异步模式的其他用户大致相同，还有很多工作要做。

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 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() 时，它将尝试将所有排队的项目分派到排队的回调中。这样就不会创建只是坐在那里睡觉，等待发出信号的线程。

那个锁有点难看，但是你可以在不加锁的情况下在多个线程上排队和订阅。我想不出一种方法来做相当于只有在另一个队列上有可用的东西时才出队而不使用那个锁。

注意：我只对此进行了最低限度的测试，有几个线程。