第一个函数设计成使linq能够安全地并行执行lambda函数(即使是异步void函数)。
因此您可以执行collection.asParallel().forAllAsync(asyncx=>awaitx.action)。
第二个函数的设计目的是使您能够并行组合和执行多个IAsynceNumerable,并尽可能快地返回它们的结果。
我有以下代码:
public static async Task ForAllAsync<TSource>(
this ParallelQuery<TSource> source,
Func<TSource, Task> selector,
int? maxDegreeOfParallelism = null)
{
int maxAsyncThreadCount = maxDegreeOfParallelism ?? Math.Min(System.Environment.ProcessorCount, 128);
using SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);
IEnumerable<Task> tasks = source.Select(async input =>
{
await throttler.WaitAsync().ConfigureAwait(false);
try
{
await selector(input).ConfigureAwait(false);
}
finally
{
throttler.Release();
}
});
await Task.WhenAll(tasks).ConfigureAwait(true);
}
public static async IAsyncEnumerable<T> ForAllAsync<TSource, T>(
this ParallelQuery<TSource> source,
Func<TSource, IAsyncEnumerable<T>> selector,
int? maxDegreeOfParallelism = null,
[EnumeratorCancellation]CancellationToken cancellationToken = default)
where T : new()
{
IEnumerable<(IAsyncEnumerator<T>, bool)> enumerators =
source.Select(x => (selector.Invoke(x).GetAsyncEnumerator(cancellationToken), true)).ToList();
while (enumerators.Any())
{
await enumerators.AsParallel()
.ForAllAsync(async e => e.Item2 = (await e.Item1.MoveNextAsync()), maxDegreeOfParallelism)
.ConfigureAwait(false);
foreach (var enumerator in enumerators)
{
yield return enumerator.Item1.Current;
}
enumerators = enumerators.Where(e => e.Item2);
}
}
如果我从第二个函数中删除“toList()”,yield return将开始返回null,因为Enumerator.item1。尽管Enumerator.item2(MoveNextAsync()的结果)为true,但Current倾向于为null。
为什么?
这是一个经典的缓期执行案例。每次在未物化的IEnumerable<>
上调用求值方法时,它都要物化IEnumerable。在本例中,这是重新调用选择器并创建等待GetAsyncEnumerator调用的任务的新实例。
通过调用.toList()
可以实现IEnumerable。如果没有它,物化将与对.any()
、对forAllAsync()
的调用以及您的foreach
循环一起发生。
同样的行为可以像这样最低限度地再现:
var enumerable = new[] { 1 }.Select(_ => Task.Delay(10));
await Task.WhenAll(enumerable);
Console.WriteLine(enumerable.First().IsCompleted); // False
enumerable = enumerable.ToList();
await Task.WhenAll(enumerable);
Console.WriteLine(enumerable.First().IsCompleted); // True
在对enumerable.first()
的第一次调用中,我们得到的任务实例与我们在它之前等待的任务实例不同。
在第二个调用中,我们使用了相同的实例,因为任务已经物化为一个列表。