如何在C#类库（.NET 6.0）中限制请求频率适配HubSpot API限额

针对HubSpot的15次/秒、100次/10秒滚动限流规则，你可以在类库中实现一个线程安全的本地速率限制器，通过跟踪请求时间窗口来自动延迟请求，避免触发限流报错。

核心实现思路 #

• 维护两个滚动时间窗口的请求时间记录：1秒窗口和10秒窗口
• 每次发起API请求前，清理窗口内的过期请求记录
• 检查当前窗口内的请求数是否超过限额，计算需要等待的时间
• 等待后再执行API请求

具体代码实现 #

1. 线程安全的速率限制器类 #

using System.Collections.Concurrent;
using System.Diagnostics;

public class HubSpotRateLimiter
{
    private readonly ConcurrentQueue<DateTimeOffset> _requestTimestamps = new();
    private readonly SemaphoreSlim _semaphore = new(1, 1);
    private readonly int _maxPerSecond = 15;
    private readonly int _maxPerTenSeconds = 100;
    private readonly TimeSpan _oneSecond = TimeSpan.FromSeconds(1);
    private readonly TimeSpan _tenSeconds = TimeSpan.FromSeconds(10);

    public async Task WaitForAvailableSlotAsync(CancellationToken cancellationToken = default)
    {
        await _semaphore.WaitAsync(cancellationToken);
        try
        {
            var now = DateTimeOffset.UtcNow;
            // 清理10秒窗口外的过期记录（同时覆盖1秒窗口的过期数据）
            while (_requestTimestamps.TryPeek(out var timestamp) && now - timestamp > _tenSeconds)
            {
                _requestTimestamps.TryDequeue(out _);
            }

            // 检查10秒窗口的请求数
            while (_requestTimestamps.Count >= _maxPerTenSeconds)
            {
                var oldestTimestamp = _requestTimestamps.Peek();
                var waitTime = oldestTimestamp + _tenSeconds - now;
                if (waitTime > TimeSpan.Zero)
                {
                    await Task.Delay(waitTime, cancellationToken);
                    now = DateTimeOffset.UtcNow;
                    // 再次清理过期记录
                    while (_requestTimestamps.TryPeek(out var ts) && now - ts > _tenSeconds)
                    {
                        _requestTimestamps.TryDequeue(out _);
                    }
                }
                else
                {
                    _requestTimestamps.TryDequeue(out _);
                }
            }

            // 检查1秒窗口的请求数
            var oneSecondAgo = now - _oneSecond;
            var countInLastSecond = _requestTimestamps.Count(ts => ts >= oneSecondAgo);
            while (countInLastSecond >= _maxPerSecond)
            {
                var oldestInSecond = _requestTimestamps.First(ts => ts >= oneSecondAgo);
                var waitTime = oldestInSecond + _oneSecond - now;
                if (waitTime > TimeSpan.Zero)
                {
                    await Task.Delay(waitTime, cancellationToken);
                    now = DateTimeOffset.UtcNow;
                    oneSecondAgo = now - _oneSecond;
                    countInLastSecond = _requestTimestamps.Count(ts => ts >= oneSecondAgo);
                }
                else
                {
                    _requestTimestamps.TryDequeue(out _);
                    countInLastSecond--;
                }
            }

            // 记录当前请求时间
            _requestTimestamps.Enqueue(now);
        }
        finally
        {
            _semaphore.Release();
        }
    }
}

2. 集成到HubSpot API调用方法 #

在你的API调用类中注入或实例化这个速率限制器，每次请求前先等待可用的请求槽：

public class HubSpotApiClient
{
    private readonly HttpClient _httpClient;
    private readonly HubSpotRateLimiter _rateLimiter;

    public HubSpotApiClient(HttpClient httpClient)
    {
        _httpClient = httpClient;
        _rateLimiter = new HubSpotRateLimiter();
    }

    public async Task<T> GetAsync<T>(string endpoint, CancellationToken cancellationToken = default)
    {
        // 等待速率限制允许请求
        await _rateLimiter.WaitForAvailableSlotAsync(cancellationToken);
        
        var response = await _httpClient.GetAsync(endpoint, cancellationToken);
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<T>(cancellationToken: cancellationToken);
    }

    public async Task<TResponse> PostAsync<TRequest, TResponse>(string endpoint, TRequest requestBody, CancellationToken cancellationToken = default)
    {
        await _rateLimiter.WaitForAvailableSlotAsync(cancellationToken);
        
        var response = await _httpClient.PostAsJsonAsync(endpoint, requestBody, cancellationToken);
        response.EnsureSuccessStatusCode();
        return await response.Content.ReadFromJsonAsync<TResponse>(cancellationToken: cancellationToken);
    }
}

额外注意事项 #

• 线程安全：使用ConcurrentQueue和SemaphoreSlim确保多线程环境下的正确性，适合类库被多个线程调用的场景
• 兜底重试：即使做了提前限流，仍可能因为网络延迟或HubSpot端的计数偏差触发限流，建议在API调用中添加重试逻辑，针对429 Too Many Requests错误提取Retry-After头进行延迟重试
• 性能优化：可以优化1秒窗口的计数逻辑，比如维护一个单独的队列或计数器，避免每次遍历整个队列
• 配置化：可以将限流参数（15/秒、100/10秒）做成可配置项，方便后续调整

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