基于可等待任务的队列

Question

我想知道是否存在类似于BlockingCollection的ConcurrentQueue实现/包装器，在这种情况下，从集合中获取数据不会阻塞，而是异步的，并且会导致异步等待，直到将项放入队列。

我已经提出了我自己的实现，但它似乎没有像预期的那样执行。我在想，我是不是在重新创造已经存在的东西。

下面是我的实现：

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)
                {
                    waitingQueue.TryDequeue(out tcs);
                    queue.TryDequeue(out firstItem);
                }
            }
            if (!ok) break;
            tcs.SetResult(firstItem);
        }
    }
}

Answer 1

我不知道有什么无锁的解决方案，但你可以看看新的Dataflow library，它是Async CTP的一部分。一个简单的BufferBlock<T>应该足够了，例如：

BufferBlock<int> buffer = new BufferBlock<int>();

生产和消费最容易通过数据流块类型上的扩展方法来完成。

生产很简单，如下所示：

buffer.Post(13);

而消费是异步就绪的：

int item = await buffer.ReceiveAsync();

如果可能的话，我建议您使用Dataflow；使这样的缓冲区既高效又正确比它最初看起来要困难得多。

Answer 2

使用C# 8.0 IAsyncEnumerable和Dataflow library的简单方法

// Instatiate an async queue
var queue = new AsyncQueue<int>();

// Then, loop through the elements of queue.
// This loop won't stop until it is canceled or broken out of
// (for that, use queue.WithCancellation(..) or break;)
await foreach(int i in queue) {
    // Writes a line as soon as some other Task calls queue.Enqueue(..)
    Console.WriteLine(i);
}

AsyncQueue的实现如下：

public class AsyncQueue<T> : IAsyncEnumerable<T>
{
    private readonly SemaphoreSlim _enumerationSemaphore = new SemaphoreSlim(1);
    private readonly BufferBlock<T> _bufferBlock = new BufferBlock<T>();

    public void Enqueue(T item) =>
        _bufferBlock.Post(item);

    public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken token = default)
    {
        // We lock this so we only ever enumerate once at a time.
        // That way we ensure all items are returned in a continuous
        // fashion with no 'holes' in the data when two foreach compete.
        await _enumerationSemaphore.WaitAsync();
        try {
            // Return new elements until cancellationToken is triggered.
            while (true) {
                // Make sure to throw on cancellation so the Task will transfer into a canceled state
                token.ThrowIfCancellationRequested();
                yield return await _bufferBlock.ReceiveAsync(token);
            }
        } finally {
            _enumerationSemaphore.Release();
        }

    }
}

Answer 3

现在有一种官方方法可以做到这一点：System.Threading.Channels。它内置于.NET Core3.0及更高版本(包括.NET 5.0和6.0)的核心运行时中，但在.NET标准2.0和2.1上也可以作为NuGet包使用。您可以通读文档here。

var channel = System.Threading.Channels.Channel.CreateUnbounded<int>();

要将工作入队，请执行以下操作：

// This will succeed and finish synchronously if the channel is unbounded.
channel.Writer.TryWrite(42);

要完成通道，请执行以下操作：

channel.Writer.TryComplete();

要从通道读取，请执行以下操作：

var i = await channel.Reader.ReadAsync();

或者，如果您使用的是.NET Core3.0或更高版本：

await foreach (int i in channel.Reader.ReadAllAsync())
{
    // whatever processing on i...
}