第一个函数旨在使 linq 能够安全地并行执行 lambda 函数(甚至是 async void 函数)。
所以你可以做 collection.AsParallel().ForAllASync(async x => await x.Action)。
第二个函数旨在使您能够并行组合和执行多个 IAsyncEnumerables 并尽快返回它们的结果。
我有以下代码:
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()”,尽管 enumerator.Item2(MoveNextAsync() 的结果)为真,但由于 enumerator.Item1.Current 往往为空,yield return 开始返回 null。
为什么?