0

第一个函数旨在使 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);
        }
    }

在迭代器到达末尾后，代码会以某种方式继续返回结果。

我正在使用这些函数来组合多个调用 API 端点的 IAsyncEnumerable 函数线程，但相同类型的结果除外。

为什么？

4

1 回答 1

1

类型(IAsyncEnumerator<T>, bool)是类型的简写ValueTuple<IAsyncEnumerator<T>, bool>，它是一个值类型。这意味着在分配时它不是通过引用传递的，而是被复制的。所以这个 lambda 不能按预期工作：

async e => e.Item2 = (await e.Item1.MoveNextAsync())

它不会更改bool存储在列表中的条目部分，而是更改临时副本的值，因此不会保留更改。

要使其按预期工作，您必须切换到引用类型 元组( 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>该类不是线程安全的，因此为了同时从多个线程安全地更新它，您必须使用lock.

于 2021-04-22T01:10:17.227 回答