在 API 请求过程中使用并行 foreach？

Question

在处理请求时使用Parallel.ForEach 是否正确？我问这个是因为 async task 旨在尽可能多地参加会议，而不是尽可能快，这是 Parallel.ForEach 会做的。

简单示例：

public async Task<OperationResult> ProcessApiRequest(List<string> ids)
{
    Parallel.ForEach(ids, async (id) =>
            {         
                await this.doStuff(id);
                await this.doAnotherStuff(id);
            });

    return OperationResult.Success();
}

想象一下，我可以收到 1 个或 100 万个 id，并且我希望尽可能多地参与请求。由于我的线程将忙于处理 100 万个 id，它将难以处理与会者的新请求，对吗？

谢谢！

Answer 1

不，这是不正确的。 Parallel.ForEach 用于数据并行。它将创建与机器上的核心一样多的工作任务，对输入数据进行分区并每个分区使用一个工作人员。它对async 操作一无所知，这意味着您的代码本质上是：

Parallel.ForEach(ids, async void (int id) =>
        {         
            await this.doStuff(id);
            await this.doAnotherStuff(id);
        });

在四台机器上，这将触发 100 万个请求，一次 4 个，无需等待任何一个。它可以在任何请求有机会完成之前轻松返回。

如果您想以受控方式执行多个请求，您可以使用具有特定并行度的 ActionBlock，例如：

var options=new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 10,
    BoundedCapacity=100
}
var block=new ActionBlock<string>(async id=>{....},options);


foreach(var id in ids)
{
    await block.SendAsync(id);
}
block.Complete();
await block.Completion;

该块将处理多达 10 个并发请求。如果动作真的是异步的，或者异步等待时间很长，我们可以轻松地使用比可用内核数更高的 DOP。

输入消息是缓冲的，这意味着我们最终可能会在慢块的输入缓冲区中等待 1M 请求。为避免这种情况，BoundedCapacity 设置将在 SendAsync 无法接受更多输入时阻止。

最后，对Complete() 的调用告诉块我们已经完成，它应该处理其输入缓冲区中的所有剩余消息。我们等待他们完成await block.Completion

Answer 2

您的担心是对的，Parallel.ForEach 默认情况下会使用线程池中尽可能多的线程，线程池将逐渐扩大到它需要的最大线程数。 Task.Run 对于 Web 服务器来说通常是个坏主意，Parallel.ForEach 通常会差很多倍。

特别是考虑到ids 是无限的，您可能很快就会遇到这样一种情况，即您的请求将被排队，因为所有线程都忙于满足少数请求。

所以你的担心是对的，这种代码正在优化单个请求的延迟以实现非常低的规模，但在规模上会牺牲一个公平且性能良好的网络服务器，最终消除延迟初始延迟的胜利，并创建你是一个更广泛的服务问题。

更新 - 正如 Panagiotis Kanavos 在 cmets 中指出的那样，Parallel.ForEach 没有 Task 重载，因此只会运行委托的初始同步部分，而留下大部分异步工作排队，您的 API 刚刚着火，可能在不知不觉中忘记了。

对于使用 ChannelReader 和 ChannelWriter 的完全异步生产者消费者模式的替代版本，以及一些新的 C# 8.0 语法，您可以试试这个：

public async Task<OperationResult> ProcessApiRequest(List<string> ids)
{
    var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(100) {SingleWriter = true});

    foreach (var id in ids)
    {
        await channel.Writer.WriteAsync(id); // If the back pressure exceeds 100 ids, we asynchronously wait here
    }
    channel.Writer.Complete();

    for (var i = 0; i < 8; i++) // 8 concurrent readers
    {
        _ = Task.Run(async () =>
        {
            await foreach (var id in channel.Reader.ReadAllAsync())
            {
                await this.doStuff(id);
                await this.doAnotherStuff(id);
            }
        });
    }

    return OperationResult.Success();
}