异步等待在linq选择

我需要修改一个现有的程序,它包含以下代码:

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

但这对我来说似乎很奇怪,首先在select中使用async和__abc1。根据斯蒂芬·克利里的这个答案,我应该能够放弃这些。

然后是第二个Select,它选择结果。这是否意味着任务根本不是异步的,而是同步执行的(如此多的努力是徒劳的),或者任务将异步执行,当它完成时,执行查询的其余部分?

根据斯蒂芬·克利里的另一个回答,我是否应该像下面这样编写上面的代码:

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个代码样本做完全相同的事情?

processevensasync看起来像这样:

async Task<InputResult> ProcessEventAsync(InputEvent ev) {...}
249304 次浏览

现有的代码可以工作,但是阻塞了线程。

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

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

.Select(t => t.Result)

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

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

对第一个代码只做一个注释。这条线

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

将生成一个Task<TResult[]>所以变量应该用单数来命名。

最后,您的最后一个代码使相同的,但更简洁。

参考:的任务。等待 / 的任务。WhenAll

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中包装异常。

在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版本能够提供更优雅的工具来处理任务的集合和集合的任务。

我更喜欢这个扩展方法:

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()

我使用了下面的代码:

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)建议的解决方案。

“你能做并不意味着你应该做”;

你可以在LINQ表达式中使用async/await,这样它就会像你想要的那样运行,但是其他阅读你的代码的开发人员还能理解它的行为和意图吗?

(特别是:异步操作应该并行运行还是故意顺序运行?最初的开发者是否考虑过这一点?)

这个问题也清楚地显示了这一点,它似乎是由一个试图理解别人的代码的开发人员提出的,而不知道它的意图。为了确保这种情况不会再次发生,如果可能的话,最好将LINQ表达式重写为循环语句。

我想调用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();

我知道任务。当我们可以并行运行时,WhenAll就是我们要走的路。

我有同样的问题@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();