如何向x个客户端并发发送n个并行请求并流式返回结果？

你的需求很明确：对按TKey分组的客户端请求，要实现两层并行控制——最多同时处理x个客户端，且每个客户端内部最多并行n个请求，同时通过IAsyncEnumerable实时返回响应结果。

先看你现有实现的几个关键问题：

• 并行参数搞反了：外部TransformManyBlock的MaxDegreeOfParallelism设为n（应该是客户端并行数x），内部TransformBlock设为x（应该是单客户端请求并行数n），完全不符合需求。
• 同步枚举异步结果：内部返回blockEntry.ReceiveAllAsync().ToEnumerable()会阻塞线程，违背异步设计初衷。
• 不必要的source.Any()：会提前枚举整个输入集合，浪费性能，空集合直接yield break即可。
• Post调用未处理失败情况：如果块已取消或完成，Post会返回false，可能导致请求丢失。

优化后的TPL Dataflow实现 #

public static async IAsyncEnumerable<TOut> ParallelForEveryEntry<TKey, TSource, TOut>(
    this IEnumerable<TSource> source,
    Func<TSource, TKey> keySelector,
    int maxClientParallelism, // 对应需求中的x：同时处理的客户端数量
    int maxRequestsPerClient, // 对应需求中的n：单个客户端的并行请求数
    Func<TSource, Task<TOut>> callback,
    [EnumeratorCancellation] CancellationToken token = default)
where TKey : notnull
{
    var clientGroups = source.ToLookup(keySelector);
    if (!clientGroups.Any()) yield break;

    // 外部块：控制同时处理的客户端数量
    var clientProcessingBlock = new TransformManyBlock<IGrouping<TKey, TSource>, TOut>(
        async clientGroup =>
        {
            // 内部块：控制单个客户端的并行请求数
            var requestBlock = new TransformBlock<TSource, TOut>(
                callback,
                new ExecutionDataflowBlockOptions
                {
                    CancellationToken = token,
                    MaxDegreeOfParallelism = maxRequestsPerClient,
                    EnsureOrdered = false
                });

            // 异步发送所有请求到内部块，处理发送失败场景
            foreach (var request in clientGroup)
            {
                if (!await requestBlock.SendAsync(request, token))
                {
                    token.ThrowIfCancellationRequested();
                    throw new InvalidOperationException("Failed to send request to processing block.");
                }
            }
            requestBlock.Complete();

            // 异步枚举内部块结果，实时输出
            await foreach (var result in requestBlock.ReceiveAllAsync(token))
            {
                yield return result;
            }
        },
        new ExecutionDataflowBlockOptions
        {
            CancellationToken = token,
            MaxDegreeOfParallelism = maxClientParallelism,
            EnsureOrdered = false
        });

    // 异步发送所有客户端分组到外部块
    foreach (var group in clientGroups)
    {
        if (!await clientProcessingBlock.SendAsync(group, token))
        {
            token.ThrowIfCancellationRequested();
            throw new InvalidOperationException("Failed to send client group to processing block.");
        }
    }
    clientProcessingBlock.Complete();

    // 实时读取外部块输出返回给调用方
    await foreach (var result in clientProcessingBlock.ReceiveAllAsync(token))
    {
        yield return result;
    }

    // 等待所有块完成，传播异常
    await clientProcessingBlock.Completion;
}

关键改进点 #

1. 参数语义化：把易混淆的n和x改为maxClientParallelism和maxRequestsPerClient，明确对应需求层级。
2. 修正并行控制逻辑：
   • 外部块控制客户端并行数，内部块控制单客户端请求并行数，完全匹配需求。
3. 异步流优化：
   • 使用SendAsync替代Post，确保在块无法接收数据时能正确处理取消或异常场景。
   • 内部直接返回IAsyncEnumerable，避免同步枚举阻塞线程，真正实现响应式结果返回。
4. 取消令牌强化：
   • 添加[EnumeratorCancellation]特性，支持调用方通过await foreach(...).WithCancellation(token)主动取消枚举。
   • 所有异步操作都传入取消令牌，确保信号能及时传递到所有层级。
5. 移除冗余操作：去掉source.Any()提前枚举逻辑，空分组自然不会进入后续处理流程。

轻量替代方案：无需TPL Dataflow #

如果不想依赖TPL Dataflow，可通过Parallel.ForEachAsync结合SemaphoreSlim实现两层并行控制：

public static async IAsyncEnumerable<TOut> ParallelForEveryEntry<TKey, TSource, TOut>(
    this IEnumerable<TSource> source,
    Func<TSource, TKey> keySelector,
    int maxClientParallelism,
    int maxRequestsPerClient,
    Func<TSource, Task<TOut>> callback,
    [EnumeratorCancellation] CancellationToken token = default)
where TKey : notnull
{
    var clientGroups = source.ToLookup(keySelector);
    if (!clientGroups.Any()) yield break;

    var clientSemaphore = new SemaphoreSlim(maxClientParallelism);
    var resultsQueue = new ConcurrentQueue<TOut>();
    var completionTask = Task.Run(async () =>
    {
        await Parallel.ForEachAsync(clientGroups, new ParallelOptions
        {
            MaxDegreeOfParallelism = maxClientParallelism,
            CancellationToken = token
        }, async (clientGroup, ct) =>
        {
            await clientSemaphore.WaitAsync(ct);
            try
            {
                var requestSemaphore = new SemaphoreSlim(maxRequestsPerClient);
                await Parallel.ForEachAsync(clientGroup, new ParallelOptions
                {
                    MaxDegreeOfParallelism = maxRequestsPerClient,
                    CancellationToken = ct
                }, async (request, innerCt) =>
                {
                    await requestSemaphore.WaitAsync(innerCt);
                    try
                    {
                        var result = await callback(request);
                        resultsQueue.Enqueue(result);
                    }
                    finally
                    {
                        requestSemaphore.Release();
                    }
                });
            }
            finally
            {
                clientSemaphore.Release();
            }
        });
    }, token);

    // 实时读取结果队列
    while (!completionTask.IsCompleted || resultsQueue.Count > 0)
    {
        while (resultsQueue.TryDequeue(out var result))
        {
            yield return result;
        }
        await Task.Delay(100, token); // 短暂等待避免空轮询
    }

    await completionTask;
}

该方案更轻量，但TPL Dataflow实现内置了流控制和异常传播机制，代码更简洁易维护，推荐优先使用。

内容的提问来源于stack exchange，提问作者André Nøbbe