如何解决结合 IAsyncEnumerator 并异步执行它们
第一个函数旨在使 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
保护它。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。