IAsyncEnumerator.Current returns 当枚举器集合未转换为列表时为 null
IAsyncEnumerator.Current returns null when enumerators collection is not casted to a List
第一个函数旨在使 linq 能够安全地并行执行 lambda 函数(即使是 async void 函数)。
所以你可以做 collection.AsParallel().ForAllASync(async x => await x.Action).
第二个函数旨在使您能够并行组合和执行多个 IAsyncEnumerables,并return尽快获得它们的结果。
我有以下代码:
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 开始为 return null 作为枚举数。Item1.Current 往往为 null,尽管有 enumerator.Item2 (MoveNextAsync() 的结果)为真。
为什么?
这是延迟执行的经典案例。每次您在非具体化 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()
时,我们得到的任务实例与我们在它之前的行中等待的任务实例不同。
在第二次调用中,我们使用相同的实例,因为任务已经具体化到列表中。
第一个函数旨在使 linq 能够安全地并行执行 lambda 函数(即使是 async void 函数)。
所以你可以做 collection.AsParallel().ForAllASync(async x => await x.Action).
第二个函数旨在使您能够并行组合和执行多个 IAsyncEnumerables,并return尽快获得它们的结果。
我有以下代码:
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 开始为 return null 作为枚举数。Item1.Current 往往为 null,尽管有 enumerator.Item2 (MoveNextAsync() 的结果)为真。
为什么?
这是延迟执行的经典案例。每次您在非具体化 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()
时,我们得到的任务实例与我们在它之前的行中等待的任务实例不同。
在第二次调用中,我们使用相同的实例,因为任务已经具体化到列表中。