如何在 Linq 查询中等待方法答案

【问题标题】：How to await a method in a Linq query如何在 Linq 查询中等待方法
【发布时间】：2013-05-13 13:01:53
【问题描述】：

尝试在 LINQ 查询中使用 await 关键字，我得到了：

“await”运算符只能用于初始“from”子句的第一个集合表达式或“join”子句的集合表达式中的查询表达式

示例代码：

var data = (from id in ids
            let d = await LoadDataAsync(id)
            select d);

难道不能在 LINQ 查询中等待某些内容，还是需要以不同的方式构造？

【问题讨论】：

我想这里涉及到太多的编译器魔法，你需要以不同的方式构造它，把它写成一个普通的 foreach 循环。

标签： linq windows-runtime async-await c#-5.0 winrt-async

【解决方案1】：

LINQ 对 async/await 的支持非常有限。对于 LINQ-to-objects，我所知道的唯一真正有用的操作是使用 async 委托执行 Select（这会导致一系列任务）。

List<T> data = new List<T>();
foreach (var id in ids)
  data.Add(await LoadDataAsync(id));

如果您可以安全地并行执行LoadDataAsync，您的示例可以重写为：

T[] data = await Task.WhenAll(ids.Select(id => LoadDataAsync(id)));

【讨论】：

这会比 Parallel.ForEach 循环执行得更好吗？
如果LoadData 是 I/O 绑定的。
太棒了！ LoadData 是一个 I/O 过程。
我编辑将LoadData重命名为LoadDataAsync，然后更清楚地表明，如果没有await，此方法将返回Task，因此WhenAll方法将起作用。跨度>
@StephenCleary Select 有什么特别之处，“我所知道的唯一真正有用的操作是使用异步委托执行 Select” 而不是 Where？

【解决方案2】：

您可以自己定义一些异步 linq 操作（用于 linq to objects）：例如：您可以编写自己的 WhereAsync 扩展方法：

public static async Task<IEnumerable<T>> WhereAsync<T>(
this IEnumerable<T> target, Func<T, Task<bool>> predicateAsync)
{
   var tasks = target.Select(async x => new { Predicate = await predicateAsync(x).ConfigureAwait(false), Value = x }).ToArray();
   var results = await Task.WhenAll(tasks).ConfigureAwait(false);

   return results.Where(x => x.Predicate).Select(x => x.Value);
}

然后像这样使用它：

var ints = new List<int> { 1, 2, 3 };
var smallInts = await ints.WhereAsync(IsSmallIntAsync);

【讨论】：

这种方法以 prone to deadlock 的方式使用异步同步（正如我在博客中解释的那样）。
@StephenCleary 是的，我明白了，我讨厌 ConfigureAwait 代码，不知道为什么默认值确实使用上下文。我修好了。
它仍然容易出现死锁，除非用户总是在他们的谓词中使用ConfigureAwait。并且只使用使用 ConfigureAwait 的库（即 HttpClient 不在某些平台上）。
@StephenCleary 我在访问结果之前确实等待了任务，我认为这会将 task.Result 的行为更改为非阻塞。我如何等待所有任务并在之后访问结果？
对不起，我完全错过了第一个await。我还是建议你使用await来检索结果以避免AggregateException，但是代码不存在sync-over-async的问题。

【解决方案3】：

使用响应式扩展，可以像这样异步处理 linq 查询的结果：

(from d in ids
select LoadDataAsync(d).ToObservable()).Merge()

这为您提供了一个可观察的流，您可以通过各种方式对其进行响应。例如，您可以将结果 .Buffer 到一个带有超时的列表中。

上述实质上是说“对于 ids 中的每个 d，对其应用一个异步函数，该函数为每个 d 产生一个任务，并将其视为单个结果的可观察对象（ToObservable），并将所有这些可观察对象一起处理作为单个可观察流（合并）

【讨论】：