基于资源占用强度扩展SemaphoreSlim实现加权限流

最佳实践 #

• 动态校准权重：不要固定单个URL的权重值，定期（比如每100次调用或每小时）用滑动窗口统计最近N次调用的平均往返时间，更新权重。外部API响应时间可能波动，这样能避免单次异常值导致权重失真。
• 标准化权重粒度：把请求耗时转成整数权重单位（比如以1ms为基准，2ms请求权重为2，2000ms请求权重为2000），总许可量直接对应你设定的总并行耗时上限（比如10000对应10秒）。
• 避免请求饥饿：如果有不同权重的请求排队，要么严格按FIFO处理，要么优先放行小权重请求，根据业务需求选择策略，防止大权重请求长期占用资源导致小请求无法执行。
• 降级兜底：当请求等待权重超时或无法获取足够许可时，直接返回降级结果（比如缓存数据、默认值），避免线程池耗尽或请求整体超时。

支持该特性的第三方库 #

• Polly：其RateLimiter策略支持自定义带权重的令牌桶逻辑，可配置每个请求根据权重消耗对应数量的令牌，总令牌数对应你设定的总并行耗时上限。
• System.Threading.RateLimiting：.NET 7+内置的限流库，其中TokenBucketRateLimiter允许调用AcquireAsync时指定获取的令牌数量，完美匹配权重限流需求。只需把总令牌容量设为总并行耗时（如10000），每个请求按耗时权重申请令牌即可。

简单实现思路 #

如果不想引入第三方库，可以自己封装一个带权重的信号量，核心是维护当前已占用的权重总和与等待队列：

public class WeightedSemaphore
{
    private readonly int _maxTotalWeight;
    private readonly Queue<(TaskCompletionSource<bool>, int)> _waitQueue = new();
    private int _currentWeight;
    private readonly object _lockObj = new();

    public WeightedSemaphore(int maxTotalWeight)
    {
        _maxTotalWeight = maxTotalWeight;
    }

    public async Task WaitAsync(int weight, CancellationToken cancellationToken = default)
    {
        if (weight > _maxTotalWeight)
            throw new ArgumentOutOfRangeException(nameof(weight), "请求权重不能超过总上限");

        lock (_lockObj)
        {
            if (_currentWeight + weight <= _maxTotalWeight)
            {
                _currentWeight += weight;
                return;
            }

            var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
            _waitQueue.Enqueue((tcs, weight));
            cancellationToken.Register(() => tcs.TrySetCanceled());
        }

        await _waitQueue.Peek().Task;
    }

    public void Release(int weight)
    {
        lock (_lockObj)
        {
            _currentWeight -= weight;
            while (_waitQueue.Count > 0)
            {
                var (nextTcs, nextWeight) = _waitQueue.Peek();
                if (_currentWeight + nextWeight <= _maxTotalWeight)
                {
                    _currentWeight += nextWeight;
                    _waitQueue.Dequeue();
                    nextTcs.TrySetResult(true);
                }
                else
                {
                    break;
                }
            }
        }
    }
}

修改你的调用逻辑：

private readonly WeightedSemaphore _weightedSemaphore = new(10000); // 对应10秒总并行耗时上限

protected async Task<TResponse> AwaitSemaphoreAndExecuteRequest<TResponse>(Func<Task<TResponse>> executionDelegate, int requestWeight)
{
    try
    {
        await _weightedSemaphore.WaitAsync(requestWeight);
        return await executionDelegate();
    }
    finally
    {
        _weightedSemaphore.Release(requestWeight);
    }
}

另外，配合一个权重统计工具来动态获取每个URL的权重：

public class RequestWeightCalculator
{
    private readonly Dictionary<string, Queue<long>> _urlResponseTimes = new();
    private readonly int _windowSize = 100; // 取最近100次调用的平均耗时作为权重

    public int GetWeight(string url)
    {
        lock (_urlResponseTimes)
        {
            if (!_urlResponseTimes.TryGetValue(url, out var times) || times.Count == 0)
                return 100; // 首次调用无数据时用默认权重

            return (int)times.Average();
        }
    }

    public void RecordResponseTime(string url, long responseTimeMs)
    {
        lock (_urlResponseTimes)
        {
            if (!_urlResponseTimes.TryGetValue(url, out var times))
            {
                times = new Queue<long>();
                _urlResponseTimes[url] = times;
            }

            times.Enqueue(responseTimeMs);
            if (times.Count > _windowSize)
                times.Dequeue();
        }
    }
}

调用前通过GetWeight获取对应URL的权重，传入限流方法即可。

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