How to await a method in a Linq query

问题:

Trying to use the await keyword in a LINQ query and I get this:

The 'await' operator may only be used in a query expression within the first collection expression of the initial 'from' clause or within the collection expression of a 'join' clause

Sample Code:

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

Is it not possible to await something in a LINQ query, or does it need to be structured a different way?

回答1:

LINQ has very limited support for async/await. For LINQ-to-objects, the only really useful operation I know of is to do a Select with an async delegate (which results in a sequence of tasks).

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

If you can do LoadDataAsync in parallel safely, your example could be rewritten as:

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

回答2:

You can define some async linq operations by yourself (for linq to objects): for example: you can write your own WhereAsync extension method:

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

And use it like that:

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

回答3:

Using reactive extensions, it is possible to handle the results of a linq query asynchronously like this:

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

This is gives you an observable stream you can respond to in various ways. For example, you can then .Buffer the results into a list with a timeout.

The above in essence says "for every d in ids, apply an asynchronous function to it, which yields a task for each d, and treat that as an observable of an individual result (ToObservable), and treat all those observables together as a single observable stream (Merge)