如何跨多并发请求实现恒定并行度的异步并行任务执行

核心思路是在全局共享的下游调用层做并发控制，而不是在单个请求的处理流程里做局部限流。.NET 异步场景下最直接的实现是使用全局单例的SemaphoreSlim作为并发闸门，保证所有下游请求都经过同一个计数器校验，从根源上限制总并行数。

具体实现步骤 #

1. 改造下游ApiClient，嵌入全局限流逻辑 #

ApiClient是所有下游调用的唯一出口，本身应该注册为单例生命周期，直接把限流逻辑放在这里可以保证所有请求（不管来自哪个上游处理流程）都受同一套规则约束：

public class ApiClient
{
    // 并发控制信号量，阈值可从配置文件读取
    private readonly SemaphoreSlim _concurrencyLimiter;
    private readonly IRestClient _restClient;

    // 初始化时指定最大并行下游调用数
    public ApiClient(int maxParallelCallCount, IRestClient restClient)
    {
        _concurrencyLimiter = new SemaphoreSlim(maxParallelCallCount, maxParallelCallCount);
        _restClient = restClient;
    }

    public async Task ExecuteAsync(Collection collection, string collectionType)
    {
        // 异步等待可用并发配额，超出阈值的调用会在这里排队，不阻塞线程
        await _concurrencyLimiter.WaitAsync();
        try
        {
            // 原有验证和请求逻辑保持不变
            var isValid = collection.Validate(collectionType);
            if (!isValid)
            {
                throw new ValidationFailedException();
            }

            var restRequest = GetRequest(collection, collectionType);
            await _restClient.ExecuteTaskAsync(restRequest);
        }
        finally
        {
            // 无论请求成功、失败还是抛异常，都必须释放配额
            _concurrencyLimiter.Release();
        }
    }
}

2. 改造ItemsProcessor，取消串行逻辑，并发发起下游调用 #

原来的foreach+await是串行执行，完全没有利用并行能力，直接改成批量生成任务后统一等待即可，不需要在这一层加任何并行度控制：

public class ItemsProcessor
{
    private readonly ApiClient _downstreamClient;
    // 通过依赖注入拿到单例的ApiClient
    public ItemsProcessor(ApiClient downstreamClient)
    {
        _downstreamClient = downstreamClient;
    }

    public async Task ExecuteAsync(Collection collection)
    {
        // 拆分所有类型的子集合，批量生成下游调用任务
        var downstreamCallTasks = collection.GroupByTypes()
            .Select(group => _downstreamClient.ExecuteAsync(group.collection, group.typeName))
            .ToArray();

        // 等待当前请求关联的所有下游调用完成
        await Task.WhenAll(downstreamCallTasks);
    }
}

关键注意事项 #

• 限流器实例必须全局唯一：绝对不要在ItemsProcessor.ExecuteAsync或者ProcessRequestAsync方法内部新建SemaphoreSlim，否则每个请求会持有独立的计数器，比如单请求限5个并行，20个并发请求就会打出100个下游调用，完全达不到全局限流的目的。如果你的架构里ApiClient不是单例，就把SemaphoreSlim单独注册为单例注入到所有ApiClient实例中。
• 必须在finally块中释放信号：如果请求抛出异常（验证失败、超时、网络错误）时没有执行Release()，会造成配额泄漏，运行时间长了可用配额会降到0，所有下游调用都会永久卡住。
• 不要用同步限流方案：非异步版Semaphore、Parallel.ForEach、lock等同步原语会阻塞线程池线程，不适合这种IO密集型的API调用场景，SemaphoreSlim.WaitAsync()是纯异步等待，没有额外的线程开销。

可选扩展 #

如果需要更复杂的限流能力，比如动态调整阈值、排队超时、取消令牌支持、按时间窗口限流，可以使用.NET 7+内置的System.Threading.RateLimiting组件中的ConcurrencyLimiter替换SemaphoreSlim，核心架构逻辑不变：所有下游调用统一经过全局共享的限流器出口。
内置ConcurrencyLimiter的简化用法示例：

// 初始化全局限流器
var limiter = new ConcurrencyLimiter(new ConcurrencyLimiterOptions
{
    PermitLimit = 20, // 最大并行数
    QueueProcessingOrder = QueueProcessingOrder.OldestFirst,
    QueueLimit = int.MaxValue // 排队队列长度，可根据业务设置
});

// 调用时获取配额
using var lease = await limiter.AcquireAsync();
if (!lease.IsAcquired)
{
    // 处理获取配额失败的逻辑，比如队列满了直接返回限流响应
}
// 执行下游调用，lease释放时自动归还配额

内容的提问来源于stack exchange，提问作者Manikanta