异步迭代器 Task<IEnumerable<T>>答案

【问题标题】：Asynchronous iterator Task<IEnumerable<T>>异步迭代器 Task<IEnumerable<T>>
【发布时间】：2014-06-11 06:51:55
【问题描述】：

我正在尝试实现一个返回迭代器的异步函数。思路如下：

    private async Task<IEnumerable<char>> TestAsync(string testString)
    {
        foreach (char c in testString.ToCharArray())
        {
            // do other work
            yield return c;
        }
    }

但是，由于Task<IEnumerable<char>> 不是迭代器接口类型，因此有错误消息指出该函数不能是迭代器块。有解决办法吗？

【问题讨论】：

请注意，这只是说 首先获取可枚举 是异步的 - 它不会使其成为任何类型的异步枚举器。你真正想做的是什么？因为我怀疑这不会实现。
检查asyncenum.codeplex.com
我想运行这个函数来处理工作线程上的流，并在 UI 线程上消耗等待的结果。
如果您想使用yield，因为您的平台没有async/await，请查看Stephen Toubs's Iterate。
这似乎很适合TPL DataFlow。

标签： c# .net asynchronous iterator clr

【解决方案1】：

听起来您可能真正在寻找类似IObservable<T> 的东西，它有点像基于推送的异步IEnumerable<T>。请参阅 Reactive Extensions, a.k.a. Rx（根据 MIT 许可的代码）（无从属关系），了解大量与 IObservable<T> 一起使用的方法，使其像 LINQ-to-Objects 等一样工作。

IEnumerable<T> 的问题在于没有什么可以真正使枚举本身异步。如果您不想添加对 Rx 的依赖（这确实让 IObservable<T> 大放异彩），这个替代方案可能对您有用：

public async Task<IEnumerable<char>> TestAsync(string testString)
{
    return GetChars(testString);
}

private static IEnumerable<char> GetChars(string testString)
{
    foreach (char c in testString.ToCharArray())
    {
        // do other work
        yield return c;
    }
}

虽然我想指出的是，在不知道异步实际上做什么的情况下，可能有更好的方法来实现您的目标。您发布的所有代码实际上都不会异步执行任何操作，我真的不知道// do other work 中的任何内容是否是异步的（在这种情况下，这不是解决您的潜在问题的方法，尽管它会使您的代码编译） .

【讨论】：

现在您可以为此使用IAsyncEnumerable。它随 C#8 .Net Core 3.0 或通过 Nuget 提供。
@joe 这实际上是我对同一个问题提出的另一个答案 :-)
能否让Update: IAsyncEnumerable 更加突出？（甚至可能将其移至答案的顶部）。
@tymtam hm，有人在这个答案上进行了编辑，尽管IAsyncEnumerable 实际上是一个完全不同的东西，我已经将它列为一个单独的答案 (stackoverflow.com/a/46198386/1083771)...我'当我有时间时，我会询问元数据，最好的办法是......
显然我实际上可以拒绝编辑，所以我刚刚做到了。我还编辑了我的另一个答案，其中包含相同的内容，包括自我最初编写答案以来已最终确定的详细信息

【解决方案2】：

A more "batteries-included" implementation 这种东西，包括语言支持，现在从 C# 8.0 开始可用。

现在，当至少使用 C# 8.0（或更高版本）和 .NET Standard 2.1（或更高版本）和/或 .NET Core 3.0（或更高版本）时，原始问题中的代码可以编写如下：

private async IAsyncEnumerable<char> TestAsync(string testString)
{
    foreach (char c in testString.ToCharArray())
    {
        // do other work, which may include "await"
        yield return c;
    }
}

【讨论】：

异步流问题的具体网址是：github.com/dotnet/csharplang/issues/43

【解决方案3】：

要详细说明以前的答案，您可以使用 Reactive Extensions 的 Observable.Create<TResult> 系列方法来做您想做的事。

这是一个例子：

var observable = Observable.Create<char>(async (observer, cancel) =>
{
    for (var i = 0; !cancel.IsCancellationRequested && i < 100; i++)
    {
        observer.OnNext(await GetCharAsync());
    }
});

以下是在 LINQPad 中使用它的方法，例如：

// Create a disposable that keeps the query running.
// This is necessary, since the observable is 100% async.
var end = Util.KeepRunning();

observable.Subscribe(
    c => Console.WriteLine(c.ToString()),
    () => end.Dispose());

【讨论】：