linq select中的异步等待

Question

我需要修改一个现有程序，它包含以下代码：

var inputs = events.Select(async ev => await ProcessEventAsync(ev))
                   .Select(t => t.Result)
                   .Where(i => i != null)
                   .ToList();

但这对我来说似乎很奇怪，首先在选择中使用async和await。根据斯蒂芬克利里的this answer，我应该可以放弃这些。

然后是第二个Select，它选择了结果。这是否意味着任务根本不是异步的并且是同步执行的（付出了这么多的努力白费了），或者任务将异步执行并且当它完成时执行其余的查询？

我是否应该按照another answer by Stephen Cleary 编写上面的代码：

var tasks = await Task.WhenAll(events.Select(ev => ProcessEventAsync(ev)));
var inputs = tasks.Where(result => result != null).ToList();

和这个完全一样吗？

var inputs = (await Task.WhenAll(events.Select(ev => ProcessEventAsync(ev))))
                                       .Where(result => result != null).ToList();

虽然我正在处理这个项目，但我想更改第一个代码示例，但我不太热衷于更改（显然可以工作）异步代码。也许我只是无所顾忌，所有 3 个代码示例都做同样的事情？

ProcessEventsAsync 看起来像这样：

async Task<InputResult> ProcessEventAsync(InputEvent ev) {...}

Answer 1

var inputs = events.Select(async ev => await ProcessEventAsync(ev))
                   .Select(t => t.Result)
                   .Where(i => i != null)
                   .ToList();

但这对我来说似乎很奇怪，首先在选择中使用 async 和 await。根据 Stephen Cleary 的这个回答，我应该可以放弃这些。

对Select 的调用有效。这两行基本相同：

events.Select(async ev => await ProcessEventAsync(ev))
events.Select(ev => ProcessEventAsync(ev))

（关于从ProcessEventAsync 抛出同步异常的方式存在细微差别，但在这段代码的上下文中，这根本不重要。）

然后是选择结果的第二个 Select。这是否意味着该任务根本不是异步的并且是同步执行的（付出了这么多的努力白费了），或者该任务将异步执行并且当它完成时执行其余的查询？

表示查询被阻塞。所以它并不是真正的异步。

分解：

var inputs = events.Select(async ev => await ProcessEventAsync(ev))

将首先为每个事件启动一个异步操作。然后这一行：

                   .Select(t => t.Result)

将等待这些操作一次完成（首先它等待第一个事件的操作，然后是下一个，然后是下一个，等等）。

这是我不关心的部分，因为它会阻塞并且还会在AggregateException 中包装任何异常。

和这个完全一样吗？

var tasks = await Task.WhenAll(events.Select(ev => ProcessEventAsync(ev)));
var inputs = tasks.Where(result => result != null).ToList();

var inputs = (await Task.WhenAll(events.Select(ev => ProcessEventAsync(ev))))
                                       .Where(result => result != null).ToList();

是的，这两个例子是等价的。它们都启动所有异步操作 (events.Select(...))，然后异步等待所有操作以任意顺序完成 (await Task.WhenAll(...))，然后继续其余的工作 (Where...)。

这两个示例都与原始代码不同。原代码是阻塞的，会将异常包装在AggregateException中。

Answer 2

现有代码正在运行，但正在阻塞线程。

.Select(async ev => await ProcessEventAsync(ev))

为每个事件创建一个新任务，但是

.Select(t => t.Result)

阻塞等待每个新任务结束的线程。

另一方面，您的代码产生相同的结果但保持异步。

仅对您的第一个代码发表评论。这一行

var tasks = await Task.WhenAll(events...

将生成单个 Task，因此该变量应以单数命名。

最后你的最后一个代码是一样的，但更简洁。

供参考：Task.Wait/Task.WhenAll

Answer 3

我使用了这个代码：

public static async Task<IEnumerable<TResult>> SelectAsync<TSource,TResult>(
    this IEnumerable<TSource> source, Func<TSource, Task<TResult>> method)
{
  return await Task.WhenAll(source.Select(async s => await method(s)));
}

像这样：

var result = await sourceEnumerable.SelectAsync(async s=>await someFunction(s,other params));

---

编辑：

有些人提出了并发问题，例如当您访问数据库时，您不能同时运行两个任务。所以这里有一个更复杂的版本，它也允许特定的并发级别：

public static async Task<IEnumerable<TResult>> SelectAsync<TSource, TResult>(
    this IEnumerable<TSource> source, Func<TSource, Task<TResult>> method,
    int concurrency = int.MaxValue)
{
    var semaphore = new SemaphoreSlim(concurrency);
    try
    {
        return await Task.WhenAll(source.Select(async s =>
        {
            try
            {
                await semaphore.WaitAsync();
                return await method(s);
            }
            finally
            {
                semaphore.Release();
            }
        }));
    } finally
    {
        semaphore.Dispose();
    }
}

没有参数，它的行为与上面更简单的版本完全相同。参数为 1 时，它将按顺序执行所有任务：

var result = await sourceEnumerable.SelectAsync(async s=>await someFunction(s,other params),1);

注意：按顺序执行任务并不意味着执行会因错误而停止！

就像使用较大的并发值或未指定参数一样，所有任务都将被执行，如果其中任何一个失败，则生成的 AggregateException 将包含抛出的异常。

如果您想一个接一个地执行任务而第一个任务失败，请尝试另一种解决方案，例如 xhafan (https://stackoverflow.com/a/64363463/379279) 建议的解决方案

Answer 4

我更喜欢这个作为扩展方法：

public static async Task<IEnumerable<T>> WhenAll<T>(this IEnumerable<Task<T>> tasks)
{
    return await Task.WhenAll(tasks);
}

这样它就可以与方法链接一起使用：

var inputs = await events
  .Select(async ev => await ProcessEventAsync(ev))
  .WhenAll()

Answer 5

使用 Linq 中可用的当前方法，它看起来很丑：

var tasks = items.Select(
    async item => new
    {
        Item = item,
        IsValid = await IsValid(item)
    });
var tuples = await Task.WhenAll(tasks);
var validItems = tuples
    .Where(p => p.IsValid)
    .Select(p => p.Item)
    .ToList();

希望 .NET 的后续版本能够提供更优雅的工具来处理任务集合和集合任务。

Answer 6

我和@KTCheek 有同样的问题，我需要它按顺序执行。但是我想我会尝试使用 IAsyncEnumerable（在 .NET Core 3 中引入）和 await foreach（在 C# 8 中引入）。这是我想出的：

public static class IEnumerableExtensions {
    public static async IAsyncEnumerable<TResult> SelectAsync<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, Task<TResult>> selector) {
        foreach (var item in source) {
            yield return await selector(item);
        }
    }
}

public static class IAsyncEnumerableExtensions {
    public static async Task<List<TSource>> ToListAsync<TSource>(this IAsyncEnumerable<TSource> source) {
        var list = new List<TSource>();

        await foreach (var item in source) {
            list.Add(item);
        }

        return list;
    }
}

这可以通过说来消费：

var inputs = await events.SelectAsync(ev => ProcessEventAsync(ev)).ToListAsync();

更新：或者，您可以添加对System.Linq.Async 的引用，然后您可以说：

var inputs = await events
    .ToAsyncEnumerable()
    .SelectAwait(async ev => await ProcessEventAsync(ev))
    .ToListAsync();

Answer 7

我想打电话给Select(...)，但要确保它按顺序运行，因为并行运行会导致一些其他并发问题，所以我最终得到了这个。 我不能调用.Result，因为它会阻塞 UI 线程。

public static class TaskExtensions
{
    public static async Task<IEnumerable<TResult>> SelectInSequenceAsync<TSource, TResult>(this IEnumerable<TSource> source, Func<TSource, Task<TResult>> asyncSelector)
    {
        var result = new List<TResult>();
        foreach (var s in source)
        {
            result.Add(await asyncSelector(s));
        }
        
        return result;
    }
}

用法：

var inputs = events.SelectInSequenceAsync(ev => ProcessEventAsync(ev))
                   .Where(i => i != null)
                   .ToList();

我知道 Task.WhenAll 是我们可以并行运行时要走的路。

Answer 8

“仅仅因为你可以并不意味着你应该。”

您可能可以在 LINQ 表达式中使用 async/await 以使其行为完全符合您的要求，但任何其他阅读您的代码的开发人员仍能理解其行为和意图吗？

（特别是：异步操作应该并行运行还是故意按顺序运行？原始开发人员是否考虑过？）

the question 也清楚地表明了这一点，这似乎是开发人员在不知道其意图的情况下试图理解别人的代码时提出的。为确保不会再次发生这种情况，最好将 LINQ 表达式重写为循环语句（如果可能）。