问题描述
第一个函数旨在使 linq 能够安全地并行执行 lambda 函数(即使是异步无效函数)。
所以你可以做 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,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);
}
}
代码在迭代器结束后以某种方式继续返回结果。
我正在使用这些函数来组合调用 API 端点的 IAsyncEnumerable 函数的多个线程,但相同类型的结果除外。
为什么?
解决方法
类型 (IAsyncEnumerator<T>,bool)
是 ValueTuple<IAsyncEnumerator<T>,bool>
类型的简写,它是一个 value type。这意味着在赋值时它不是通过引用传递的,而是被复制的。所以这个 lambda 不能按预期工作:
async e => e.Item2 = (await e.Item1.MoveNextAsync())
它不会更改存储在列表中的条目的 bool
部分,而是更改临时副本的值,因此不会保留更改。
要使其按预期工作,您必须切换到 reference type tuples (Tuple<IAsyncEnumerator<T>,bool>
),或替换列表中的整个条目:
List<(IAsyncEnumerator<T>,bool)> enumerators = source./*...*/.ToList()
//...
var entry = enumerators[index];
enumerators[index] = (entry.Item1,await entry.Item1.MoveNextAsync());
请注意 List<T>
类 is not thread-safe,因此为了从多个线程并发地安全更新它,您必须使用 lock
保护它。