在 Parallel.Foreach 中调用异步方法的最佳实践/正确方法是什么?

Question

我正在创建一个 .net 核心应用程序，我需要在其中调用async方法来处理大量对象。我们已经在Parallel.ForEach内完成了此操作，因此我们可以利用并行性更快地完成作业。

服务方法是我们无法更改的async方法。我的问题是在使用 TPL 并行时调用此方法的正确方法是什么。

这是一个简化的代码片段(我传递迭代#而不是用于演示目的的对象)，以及我们的观察结果：

CallSendAsync方法在内部向 API 发出 HTTP 请求(使用 HttpClient)。

private void ParallelFor()
{
Parallel.For(0, 100000, i =>
{
CallSendAsync(i).GetAwaiter().GetResult();
});
}

我对上述代码的问题是，它使用的是使async方法同步的GetAwaiter。但是，上述实现速度非常快。它似乎也更有效地管理系统资源。

另一方面，我有这个：

private void ParallelForWithAsync()
{
Parallel.For(0, 100000, async i =>
{
await CallSendAsync(i);
});
}

此代码有一个async/await。但是，它变得非常慢，性能会显着下降。它打开了大量的出站端口，最终HTTP请求出错。

第三，我也尝试了这个：

private void TaskWaitAll()
{
IEnumerable<int> arr = Enumerable.Range(1, 100000);
Task.WhenAll(arr.Select(CallSendAsync)).GetAwaiter().GetResult();
}

其结果也与第二个片段相似。

Answer 1

我 .net 核心应用程序，我需要调用异步方法来处理大量对象。 我们已经使用 Parallel.ForEach 完成了此操作，因此我们可以利用并行性更快地完成作业。

让我就在那里阻止你。你不会"使用并行"来"让事情更快"。并行性是并发的一种形式(一次执行多项操作)，它是一种并发形式，它使用多个线程在多核计算机上更快地处理 CPU 密集型算法。但是，您的操作根本不受 CPU 限制;它是 I/O 绑定的，这表明并行是用于此的错误技术。

如果要在基于 I/O 的处理时同时处理多个项目，则适当的解决方案是使用Task.WhenAll。

但是，它变得非常慢，性能会显着下降。 它打开了大量的出站端口，最终HTTP请求出错。

是的。如果您实际上同时发出十万个HTTP请求，这是可以预期的。请记住，IANA 临时端口少于 16k 个。对于像这样的大量请求，您可能希望将其限制为更合理的数量 - 例如一次 20 个。Parallel.For将根据 CPU 使用率、线程池中的线程数等对同步工作负载进行正确分区。若要限制异步工作，可以使用SemaphoreSlim：

private async Task TaskWaitAll()
{
var mutex = new SemaphoreSlim(20);
IEnumerable<int> arr = Enumerable.Range(1, 100000);
var tasks = arr.Select(async i =>
{
await mutex.WaitAsync();
try { await CallSendAsync(i); }
finally { mutex.Release(); }
}).ToList();
await Task.WhenAll(tasks);
}