是否有像 Queue 这样实现 IAsyncEnumerable 的 C# 类?

Is there a C# class like Queue that implements IAsyncEnumerable?(是否有像 Queue 这样实现 IAsyncEnumerable 的 C# 类?)

问题描述

Queue 和 ConcurrentQueue 都实现 IEnumerable 但不实现 IAsyncEnumerable.NuGet 上是否有实现 IAsyncEnumerable 的标准类或类，这样，如果队列为空，则 MoveNextAsync 的结果直到将下一个内容添加到排队?

Both Queue and ConcurrentQueue implement IEnumerable but not IAsyncEnumerable. Is there a standard class or class available on NuGet which implements IAsyncEnumerable such that, if the queue is empty, the result of MoveNextAsync does not complete until something next is added to the queue?

推荐答案

如果您使用的是 .NET Core 平台，则至少有两个内置选项:

If you are using the .NET Core platform there are at least two built-in options:

1. System.Threading.Tasks.Dataflow.BufferBlock<T> 类，TPL 数据流 库.它没有原生实现 IAsyncEnumerable<T>，但它公开了可等待的 OutputAvailableAsync() 方法，实现 ToAsyncEnumerable 很简单扩展方法.

1. The System.Threading.Tasks.Dataflow.BufferBlock<T> class, part of the TPL Dataflow library. It doesn't implement the IAsyncEnumerable<T> natively, but it exposes the awaitable OutputAvailableAsync() method, doing it trivial to implement a ToAsyncEnumerable extension method.

System.Threading.Channels.Channel<T> 类，频道 库.它通过其公开 IAsyncEnumerable<T> 实现Reader.ReadAllAsync()¹ 方法.

The System.Threading.Channels.Channel<T> class, the core component of the Channels library. It exposes an IAsyncEnumerable<T> implementation via its Reader.ReadAllAsync()¹ method.

通过安装 nuget 包(每个类不同)，这两个类也可用于 .NET Framework.

Both classes are also available for .NET Framework, by installing a nuget package (different for each one).

BufferBlock 的 IAsyncEnumerable 实现:

public static async IAsyncEnumerable<T> ToAsyncEnumerable<T>(
    this IReceivableSourceBlock<T> source,
    [EnumeratorCancellation]CancellationToken cancellationToken = default)
{
    while (await source.OutputAvailableAsync(cancellationToken).ConfigureAwait(false))
    {
        while (source.TryReceive(out T item))
        {
            yield return item;
            cancellationToken.ThrowIfCancellationRequested();
        }
    }
    await source.Completion.ConfigureAwait(false); // Propagate possible exception
}