c# - 从 IEnumerable> 到 IAsyncEnumerable 通过 yield 在 Parallel.ForEach/Parallel.ForEachAsync 内返回给出错误 CS1621

Question

在 .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;
}

但我在 yield 处收到以下错误:CS1621 - yield 语句不能在匿名方法或 lambda 表达式中使用。
有没有办法达到我想要的结果？
如果我不够清楚，请随时提出问题!

Answer 1

您可以为此目的使用的最容易获得的工具是 TransformBlock<TInput,TOutput> 来自 TPL 数据流库。该组件在 .NET Core 及更高版本中 native 可用，它本质上是具有两个队列(输入和输出)的处理器/投影仪/变压器。您指定处理函数，然后根据需要配置选项，然后输入数据，最后检索处理后的输出:

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);

    TransformBlock<int, ApiResponse> block = new(async page =>
    {
        return await CallApiAsync(page, cancellationToken);
    }, new ExecutionDataflowBlockOptions()
    {
        CancellationToken = cancellationToken,
        MaxDegreeOfParallelism = 10, // Configurable, the default is 1
        EnsureOrdered = true, // This is the default
    });

    // Feed the block with input data
    foreach (var page in pages) block.Post(page);
    block.Complete();

    // Emit the output data as they become available
    while (await block.OutputAvailableAsync())
        while (block.TryReceive(out var item))
            yield return item;

    // Propagate possible exception (including cancellation)
    await block.Completion;
}

这个简单的实现启动 TransformBlock当得到 IAsyncEnumerable<ApiResponse>被枚举，它不会停止，直到所有处理完成，或 cancellationToken被取消。处理不是由结果序列的枚举驱动。如果客户端代码简单地放弃枚举，它甚至不会停止，break使用 await foreach环形。如果你想包含这个功能(优雅终止)，你必须添加一个 try - finally block 和内部链接 CancellationTokenSource如图here .屈服循环应放置在 try 内。 ，并取消链接CancellationTokenSource应该放在finally里面.