标签: iasyncenumerable

我确信这个问题会证明我的无知，但我很难理解这一点。我愿意问一个愚蠢的问题以获得一个好的答案。

我读过的所有关于异步流的帖子都很好地展示了该功能，但它们没有解释为什么它比替代方案有所改进。

或者，也许，什么时候应该通过良好的旧客户端 - 服务器通信使用异步流？

我可以看到流式传输大文件的内容对于异步流来说可能是一个很好的用途，但是我看到的许多示例都使用异步流来传输少量传感器数据（例如温度）。似乎带有温度传感器的 IoT 设备只需通过 HTTP POST 将数据发送到服务器，服务器就可以响应。在这种情况下，服务器为什么要实现异步流？

当你努力理解这些话时，我已经能感受到你的痛苦，但请怜悯我。:)

根据要求，这里有一些我遇到的让我感到困惑的例子。当我找到它们时，我会发布更多，但我想继续让你开始：

• .NET Conf 主题演讲的前半部分是一个大规模的异步流演示......我不明白他们为什么在这里使用异步流：https : //www.youtube.com/watch?v= 1xQE2bWkwjo & list = PLReL099Y5nRd04p81Q7p5TtyjCrj9tz1t & index =4&t =
• 这是另一个让我困惑的例子

尝试从 ASP.NET Core 3 SignalR Hub 捕获顶级异常

这很棘手，因为我使用的是 yield return，并且您不能将其包装在 try-catch 块中。它给出了这个编译器错误：

CS1626 C# 无法在带有 catch 子句的 try 块的主体中​​产生值

在这里讨论

那么，如何捕获这个异常呢？它被困在内部某处并发送到 javascript 客户端。我似乎看不到 ASP.NET Core 中间件管道中的异常。

// SignalR Hub 
public class CrawlHub : Hub
{
    public async IAsyncEnumerable<UIMessage> Crawl(string url, [EnumeratorCancellation]CancellationToken cancellationToken)
    {
        Log.Information("Here");
        // Trying to catch this error further up pipeline as
        // can't try catch here due to yield return
        throw new HubException("This error will be sent to the client!");
        // handing off to Crawler which returns …

我们有这样的代码：

var intList = new List<int>{1,2,3};
var asyncEnumerables = intList.Select(Foo);

private async IAsyncEnumerable<int> Foo(int a)
{
  while (true)
  {
    await Task.Delay(5000);
    yield return a;
  } 
}

我需要await foreach为每个asyncEnumerable条目开始。每次循环迭代都应该相互等待，每次迭代完成后，我需要收集每次迭代的数据并通过另一种方法对其进行处理。

我可以通过 TPL 以某种方式实现吗？否则，你不能给我一些想法吗？

我有一个 .net core 3.1 控制台应用程序。

我有一个具有以下签名的方法：

public async IAsyncEnumerable<string> GetFilePathsFromRelativePathAsync(string relativePath)

如果我称之为：

private async Task<IEnumerable<FileUpload>> GetFileUploadsAsync(string relativePath)
{
...
    var filePaths = await service.GetFilePathsFromRelativePathAsync(relativePath);
...
}

我收到以下错误：

错误 CS1061“IAsyncEnumerable”不包含“GetAwaiter”的定义，并且找不到接受“IAsyncEnumerable”类型的第一个参数的可访问扩展方法“GetAwaiter”（您是否缺少 using 指令或程序集引用？）

答案可能是不可能，但问题是：假设您有一个 C# 方法来使用TextReader返回 的a 中的行IAsyncEnumerable<string>。如何确保在DisposeAsync调用时IAsyncEnumerator<string>会TextReader被处理掉？或者这是您需要编写自定义实现才能实现的目标？

When working with an IEnumerable<T> there are the build-in extension methods from the System.Linq namespace such as Skip, Where and Select to work with.

When Microsoft added IAsyncEnumerable in C#8 did they also add new Linq methods to support this?

I could of course implement these methods myself, or maybe find some package which does that, but I'd prefer to use a language-standard method if it exists.

两者Queue和ConcurrentQueue实施IEnumerable但不是IAsyncEnumerable。NuGet 上是否有可用的标准类或类来实现IAsyncEnumerable，如果队列为空，则在将MoveNextAsync下一个添加到队列之前，结果不会完成？

请参阅以下两种方法。第一个返回一个IAsyncEnumerable. 第二个尝试消耗它。

using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;

public static class SqlUtility
{
    public static async IAsyncEnumerable<IDataRecord> GetRecordsAsync(
        string connectionString, SqlParameter[] parameters, string commandText,
        [EnumeratorCancellation]CancellationToken cancellationToken)
    {
        using (SqlConnection connection = new SqlConnection(connectionString))
        {
            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
            using (SqlCommand command = new SqlCommand(commandText, connection))
            {
                command.Parameters.AddRange(parameters);
                using (var reader = await command.ExecuteReaderAsync()
                    .ConfigureAwait(false))
                {
                    while (await reader.ReadAsync().ConfigureAwait(false))
                    {
                        yield return reader;
                    }
                }
            }
        }
    }

    public static async Task Example() …

我正在尝试找出带来IAsyncEnumerable<T>诸如IEnumerable<Task<T>>.

我编写了以下类，它允许我等待一系列数字，每个数字之间有定义的延迟：

class DelayedSequence : IAsyncEnumerable<int>, IEnumerable<Task<int>> {
    readonly int _numDelays;
    readonly TimeSpan _interDelayTime;

    public DelayedSequence(int numDelays, TimeSpan interDelayTime) {
        _numDelays = numDelays;
        _interDelayTime = interDelayTime;
    }

    public IAsyncEnumerator<int> GetAsyncEnumerator(CancellationToken cancellationToken = default) {
        async IAsyncEnumerable<int> ConstructEnumerable() {
            for (var i = 0; i < _numDelays; ++i) {
                await Task.Delay(_interDelayTime, cancellationToken);
                yield return i;
            }
        }

        return ConstructEnumerable().GetAsyncEnumerator();
    }

    public IEnumerator<Task<int>> GetEnumerator() {
        IEnumerable<Task<int>> ConstructEnumerable() {
            for (var i = 0; i < _numDelays; ++i) { …

在 .NET 6 项目中，我必须调用一个偏移分页（页/每页）的 Web API，并且我希望尽可能使 n 个调用并行。

这是使用给定页码调用 API 一次的方法：

private Task<ApiResponse> CallApiAsync(int page,
    CancellationToken cancellationToken = default)
{
    return GetFromJsonAsync<ApiResponse>($"...&page={page}", cancellationToken)
        .ConfigureAwait(false);
}

我实际上需要的是从第 1 页到第 n 页的所有 API 调用的仅前向流式迭代器，因此考虑到这一要求，我认为这IAsyncEnumerable是正确的 API，这样我就可以并行触发 API 调用并访问每个 API 响应一旦准备好，就可以完成，而不需要全部完成。

所以我想出了以下代码：

public async IAsyncEnumerable<ApiResponse> CallApiEnumerableAsync(int perPage,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
    int numProducts = GetNumberOfProducts(perPage);

    int numCalls = MathExtensions.CeilDiv(numProducts, perPage);

    var pages = Enumerable.Range(1, numCalls);

    Parallel.ForEach(pages, async page => {
        yield return await CallApiAsync(page, cancellationToken).ConfigureAwait(false);
    });

    yield break;
}

但我收到以下错误 …