带有异步 lambda 的 Parallel.ForEach 等待所有迭代完成

Question

最近我看到几个与 Parallel.ForEach 相关的 SO 线程与异步 lambda 混合，但所有建议的答案都是某种解决方法。

有什么办法可以写：

List<int> list = new List<int>[]();

Parallel.ForEach(arrayValues, async (item) =>
{
  var x = await LongRunningIoOperationAsync(item);
  list.Add(x);
});

如何确保列表包含每次迭代中使用 lambdas 执行的所有迭代中的所有项目？

Parallel.ForEach 通常如何与异步 lambda 一起工作，如果它命中 await 是否会将其线程移交给下一次迭代？

我假设 ParallelLoopResult IsCompleted 字段不是正确的，因为它会在所有迭代执行时返回 true，无论它们的实际 lambda 作业是否完成？

Answer 1

最近我看到几个与 Parallel.ForEach 相关的 SO 线程与异步 lambda 混合，但所有建议的答案都是某种解决方法。

嗯，那是因为 Parallel 不适合 async。从不同的角度来看，你为什么要首先混合它们？他们做相反的事情。 Parallel 是关于添加线程，async 是关于放弃线程。如果你想同时做异步工作，那么使用Task.WhenAll。那是工作的正确工具； Parallel 不是。

也就是说，听起来你想使用错误的工具，所以这是你的做法......

如何确保列表包含每次迭代中使用 lambdas 执行的所有迭代中的所有项目？

在处理完成之前，您需要有某种信号可以阻止某些代码，例如 CountdownEvent 或 Monitor。附带说明一下，您还需要保护对非线程安全 List<T> 的访问。

Parallel.ForEach 通常如何与异步 lambda 一起工作，如果它命中 await 是否会将其线程移交给下一次迭代？

由于Parallel 不理解async lambda，当第一个await 产生（返回）给它的调用者时，Parallel 将假定循环的交互已完成。

我假设 ParallelLoopResult IsCompleted 字段不是正确的，因为它会在所有迭代执行时返回 true，无论它们的实际 lambda 作业是否完成？

正确。据Parallel 所知，它只能“看到”第一个await 返回其调用者的方法。所以它不知道async lambda 何时完成。它还会假设迭代过早完成，从而导致分区失效。

Answer 2

你不需要Parallel.For/ForEach在这里你只需要等待一个任务列表。

背景

简而言之，您需要非常小心异步 lambdas，如果您将它们传递给 Action 或 Func<Task>

您的问题是因为Parallel.For / ForEach 不适合异步和等待模式 或IO 绑定任务。它们适用于 cpu 绑定的工作负载。这意味着它们本质上具有Action 参数，让任务调度程序为您创建任务

如果您想同时运行多个 async 任务，请使用 Task.WhenAll 或 TPL 数据流 Block（或类似的东西) 可以有效地处理 CPU bound 和 IO bound 工作负载，或者更直接地说，它们可以处理 tasks 异步方法是。

除非你需要在你的 lambda 中做更多的事情（你还没有展示），否则只需使用 aSelect 和 WhenAll

var tasks = items .Select(LongRunningIoOperationAsync);
var results = await Task.WhenAll(tasks); // here is your list of int

如果你这样做了，你仍然可以使用等待，

var tasks = items.Select(async (item) =>
   {
       var x = await LongRunningIoOperationAsync(item);
       // do other stuff
       return x;
   });

var results = await Task.WhenAll(tasks);

_{注意：如果你需要Parallel.ForEach的扩展功能（即控制最大并发的选项），有几种方法，但是RX或DataFlow 可能是最简洁的}