相关疑难解决方法(0)

我想知道是否存在ConcurrentQueue的实现/包装器,类似于BlockingCollection,其中从集合中获取不会阻塞,而是异步并且将导致异步等待直到将项目放入队列中.

我已经提出了自己的实现,但它似乎没有按预期执行.我想知道我是否正在重塑已经存在的东西.

这是我的实现:

public class MessageQueue<T>
{
    ConcurrentQueue<T> queue = new ConcurrentQueue<T>();

    ConcurrentQueue<TaskCompletionSource<T>> waitingQueue = 
        new ConcurrentQueue<TaskCompletionSource<T>>();

    object queueSyncLock = new object();

    public void Enqueue(T item)
    {
        queue.Enqueue(item);
        ProcessQueues();
    }

    public async Task<T> Dequeue()
    {
        TaskCompletionSource<T> tcs = new TaskCompletionSource<T>();
        waitingQueue.Enqueue(tcs);
        ProcessQueues();
        return tcs.Task.IsCompleted ? tcs.Task.Result : await tcs.Task;
    }

    private void ProcessQueues()
    {
        TaskCompletionSource<T> tcs=null;
        T firstItem=default(T);
        while (true)
        {
            bool ok;
            lock (queueSyncLock)
            {
                ok = waitingQueue.TryPeek(out tcs) && queue.TryPeek(out firstItem);
                if (ok)
                { …

我想出了一些代码来消耗队列中所有的wating项目.而不是逐项处理项目,将所有等待项目作为一组处理是有意义的.

我已经宣布我的队列是这样的.

private BlockingCollection<Item> items = 
    new BlockingCollection<Item>(new ConcurrentQueue<Item>);

然后,在消费者线程上,我打算像这样批量阅读这些项目,

Item nextItem;
while (this.items.TryTake(out nextItem, -1))
{
    var workToDo = new List<Item>();
    workToDo.Add(nextItem);

    while(this.items.TryTake(out nextItem))
    {
        workToDo.Add(nextItem);
    }

    // process workToDo, then go back to the queue.
}

这种方法缺乏实用性,GetConsumingEnumerable我不禁想知道我是否错过了更好的方法,或者我的方法是否存在缺陷.

有没有更好的方法来BlockingCollection<T>批量消费？