Как потреблять API с ограничением по RPS в .NET приложениях

Однажды каждый C# программист получает на работе задачу по разработке интеграции с внешней системой, где ограничена максимальная частота запросов в секунду.

Интернет яростно сопротивлялся предоставить мне инструкцию к написанию такого кода, закидывая туториалами по настройке ограничения RPS на сервере, а не клиенте.

Но теперь на Хабре есть эта статья, которая научит отправлять запросы из HttpClient так, чтобы не получать 429 Too Many Requests.

Throttling

Обозначенный процесс взаимодействия с API, где нельзя превышать заданный RPS, в общем случае именуется троттлингом (throttling).

Throttling — это широко используемая техника для увеличения производительности кода, который выполняется повторно с некоторой периодичностью.

Троттлинг функции означает, что функция вызывается не более одного раза в указанный период времени (например, раз в 10 секунд). Другими словами ― троттлинг предотвращает запуск функции, если она уже запускалась недавно. Троттлинг также обеспечивает регулярность выполнения функции с заданной периодичностью.



В экшен-играх приходится нажимать кнопки с высокой частотой для выполнения какого-либо действия (стрельба, удар). Как правило, игроки нажимают кнопки намного чаще, чем это требуется, вероятно, увлекаясь происходящим. Таким образом игрок может нажать на кнопку «удара» 10 раз в течение пяти секунд, но персонаж делает не более одного удара в секунду. В этом случае троттлинг события «удар» позволяет игнорировать повторные нажатия кнопки в течение секунды.

Как реализуется подобное ограничение?

Rate Limiting Алгоритмы

Есть ряд алгоритмов, которые реализуют технику Rate Limiting — контроль количества допущенного трафика к объекту.



Существует множество различных алгоритмов ограничения скорости для управления потоком запросов. .NET 7 представил 4 таких алгоритма.

▍ Concurrency limit

Concurrency лимитер ограничивает количество одновременных запросов, которые могут получить доступ к ресурсу. Если установленный предел равен 10, то 10 запросов могут получить доступ к ресурсу одновременно, а 11-й запрос не будет допущен. Как только запрос завершается, количество разрешённых запросов увеличивается до 1, при завершении двух запросов — до 2 и так далее. Это делается с помощью вызова Dispose на экземпляре RateLimitLease, о которой поговорим позже.

▍ Token bucket limit

Token bucket алгоритм получил своё название, исходя из принципа работы. Представьте, что есть ведро, до краёв наполненное токенами. Когда поступает запрос, он забирает токен и хранит его вечно. Через некоторое время кто-то добавляет в ведро заранее определённое количество токенов, никогда не добавляя больше, чем ведро может вместить. Если ведро пустое, то при поступлении запроса ему будет отказано в доступе к ресурсу.

Приведу более конкретный пример. Предположим, что ведро вмещает 10 токенов, и каждую минуту в него добавляется 2 токена. Когда приходит запрос, он забирает 1 токен, так что у нас остаётся 9. Ещё 3 запроса приходят и забирают 3 токена, оставляя в ведре 6 токенов. Через минуту поступает 2 новых токена, что даёт 8 в сумме. 8 запросов приходят и забирают оставшиеся токены, опустошая ведро. Если приходит ещё один запрос после этого, то у него уже не получится взять доступ к ресурсу, пока в ведре не окажется больше токенов. Они в данном примере восполняются каждую минуту. Через 5 минут отсутствия запросов в ведре снова будут все 10 токенов, и в последующие минуты они не будут добавляться, пока новые запросы не начнут забирать токены.

▍ Fixed window limit

Алгоритм с фиксированным окном использует идею окна, которая будет использоваться и в следующем алгоритме. Окно — это промежуток времени, в течение которого действует ограничение, прежде чем произойдёт переход к следующему окну. В случае с фиксированным окном переход к следующему окну означает сброс ограничения обратно в начальную точку.

Представим, что есть кинотеатр с одним залом, вмещающим 100 человек, которые пришли смотреть 2-часовой фильм. Когда фильм начинается, людям разрешается встать в очередь на следующий сеанс, который состоится через 2 часа. В очереди могут стоять до 100 человек, прежде чем им начнут говорить прийти позже. По прошествии двух часов фильм заканчивается, и очередь от 0 до 100 человек может переместиться в кинотеатр, тем самым начиная формирование новой очереди. Всё равно что двигать само окно в алгоритме фиксированного окна.

▍ Sliding window limit

Алгоритм со скользящим окном похож на алгоритм с фиксированным окном. Но туда добавляются отрезки. Отрезок — это, соответственно, часть окна. Если взять 2-часовое окно из предыдущего раздела и разбить его на 4 отрезка, то получится четыре 30-минутных отрезка. Также учитывается индекс текущего отрезка, который будет всегда указывать на самый новый отрезок в окне. Запросы в течение получасового периода попадают в текущий отрезок, и каждые 30 минут окно сдвигается на один отрезок. Если в течение отрезка, мимо которого проскользнуло окно, были запросы, они обновляются, и установленное ограничение увеличивается на эту величину. Если запросов не было, наше ограничение остаётся прежним.

Время кодить

Окей, вот и разобрались немного с теоретической базой. Менеджер не дремлет и всё ещё просит задачу по интеграции, поэтому надо как можно скорее начать писать код.

Вот так, кстати, он выглядит:

Предположим, что есть некий DataObject, содержащий некий Content.

Этот DataObject можно получить, вызвав какой-нибудь IDataObjectExternalApiService, под капотом у которого экземпляр HttpClient делает запрос:

record DataObject(string Content);

interface IDataObjectExternalApiService
{
    Task<DataObject> GetByIdAsync(int id, CancellationToken ct = default);
}

Есть какой-то набор идентификаторов, исчисляющийся тысячами, и по ним нужно выкачать соответствующие контенты:

interface IDataObjectCollectionProvider
{
    Task<IReadOnlyCollection<DataObject>> GetByIdsAsync(
        IReadOnlyCollection<int> ids,
        CancellationToken ct = default);
}

Но вот незадача! Это самое внешнее API допускает максимум только 10 RPS.

▍ SemaphoreSlim

Если допускается, что речь идёт о 10 запросах одновременно, то можно попробовать реализовать Concurrency лимитер с помощью примитива синхронизации SemaphoreSlim.

SemaphoreSlim — это облегчённая альтернатива Semaphore, которая ограничивает количество потоков, имеющих одновременный доступ к ресурсу или пулу ресурсов. При этом SemaphoreSlim можно использовать в рамках async/await.

Получится примерно что-то такое:

class DataObjectCollectionProvider : IDataObjectCollectionProvider
{
    private readonly IDataObjectExternalApiService _externalApiService;

    public DataObjectCollectionProvider(IDataObjectExternalApiService externalApiService) =>
        _externalApiService = externalApiService;

    public async Task<IReadOnlyCollection<DataObject>> GetByIdsAsync(
        IReadOnlyCollection<int> ids,
        CancellationToken ct = default)
    {
        if (ids.Count == 0)
            return [];

        var semaphoreSlim = new SemaphoreSlim(
            initialCount: 10,
            maxCount: 10);
        ConcurrentBag<DataObject> dataObjects = [];

        var tasks = ids.Select(async id =>
        {
            await semaphoreSlim.WaitAsync(ct);
            try
            {
                var dataObject = await _externalApiService.GetByIdAsync(id, ct);
                dataObjects.Add(dataObject);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        });

        await Task.WhenAll(tasks);
        return dataObjects;
    }
}

На мой взгляд, выглядит достаточно кустарно — надо встраивать использование примитива синхронизации в логику вызова и перебора, перемешивая инфраструктурный код с бизнес-логикой.

А если понадобится другой алгоритм, что делать в таком случае?

▍ System.Threading.RateLimiting

Как уже было сказано, .NET 7 представил новый NuGet-пакет System.Threading.RateLimiting, содержащий реализации указанных в статье алгоритмов.

Все они являются наследниками абстрактного класса RateLimiter:

public abstract class RateLimiter : IAsyncDisposable, IDisposable
{
    public abstract int GetAvailablePermits();
    public abstract TimeSpan? IdleDuration { get; }

    public RateLimitLease Acquire(int permitCount = 1);
    public ValueTask<RateLimitLease> WaitAsync(int permitCount = 1, CancellationToken cancellationToken = default);

    public void Dispose();
    public ValueTask DisposeAsync();
}

Наследники этого класса принимают в качестве параметра специальные конфигурационные настройки для регулирования поведения алгоритма.

Соответственно, тут уже выбор какой-то появляется. Например, указанная задача, по моему скромному мнению, решается интуитивнее с помощью Fixed Window.

Тогда и переписать код можно следующим образом:

class DataObjectCollectionProvider : IDataObjectCollectionProvider
{
    private readonly IDataObjectExternalApiService _externalApiService;

    public DataObjectCollectionProvider(IDataObjectExternalApiService externalApiService) =>
        _externalApiService = externalApiService;

    public async Task<IReadOnlyCollection<DataObject>> GetByIdsAsync(
        IReadOnlyCollection<int> ids,
        CancellationToken ct = default)
    {
        if (ids.Count == 0)
            return [];

        var limiter = new FixedWindowRateLimiter(
            new FixedWindowRateLimiterOptions
            {
                Window = TimeSpan.FromSeconds(1),
                PermitLimit = 10,
                QueueLimit = 10
            });
        ConcurrentBag<DataObject> dataObjects = [];

        var tasks = ids.Select(async id =>
        {
            using var lease = await limiter.AcquireAsync(cancellationToken: ct);
            if (lease.IsAcquired)
            {
                var dataObject = await _externalApiService.GetByIdAsync(id, ct);
                dataObjects.Add(dataObject);
            }
        });

        await Task.WhenAll(tasks);
        return dataObjects;
    }
}

Уже лучше, поскольку алгоритм заменим, а опции можно внедрить через DI.

Кстати, обратите внимание, что создаётся лимитер с параметрами PermitLimit = 10 и QueueLimit = 10. Это означает, что в окно размером в секунду пустят не больше 10 запросов, и разрешается ставить в очередь вызовы WaitAsync с общим количеством запросов на разрешение не более 10.

Однако что делать, если разрешение не удастся выбить, и лимитер не пустит наш запрос? Как построить логику обработки ошибок? И всё это сделать так, чтобы можно было написать unit-тесты?

▍ Polly.RateLimiting

Тут на помощь приходит библиотека Polly, которая построила обёртку над System.Threading.RateLimiting в виде пакета Polly.RateLimiting.

Вообще, инструмент сильно изменился, добавив концепцию пайплайнов. Пайплайн оборачивает необходимый вызов, но перед этим в него накручиваются все свистелки с перделками. Подробности вот в этом разделе документации.

Так вот лимитеры тоже можно добавлять в качестве этапа пайплайна, а сам пайплайн протаскивать через DI до конкретного потребителя.

Внедрение происходит с помощью специального провайдера, который достаёт пайплайн по заданному ключу. Для простоты в качестве ключа буду использовать имя типа потребителя.

Получается, что можно задекорировать внешний сервис, обернув вызов, а затем использовать его как угодно:

class DataObjectServiceRateLimiterDecorator : IDataObjectExternalApiService
{
    private readonly IDataObjectExternalApiService _decorated;
    private readonly ResiliencePipeline<DataObject> _pipeline;

    public DataObjectServiceRateLimiterDecorator(
        IDataObjectExternalApiService decorated,
        ResiliencePipelineProvider<string> pipelineProvider)
    {
        _decorated = decorated;
        _pipeline = pipelineProvider.GetPipeline<DataObject>(
            key: nameof(DataObjectServiceRateLimiterDecorator));
    }

    public async Task<DataObject> GetByIdAsync(int id, CancellationToken ct = default) =>
        await _pipeline.ExecuteAsync(async token => await _decorated.GetByIdAsync(id, token), ct);
}

class DataObjectCollectionProvider : IDataObjectCollectionProvider
{
    private readonly IDataObjectExternalApiService _externalApiService;

    public DataObjectCollectionProvider(IDataObjectExternalApiService externalApiService) =>
        _externalApiService = externalApiService;

    public async Task<IReadOnlyCollection<DataObject>> GetByIdsAsync(
        IReadOnlyCollection<int> ids,
        CancellationToken ct = default)
    {
        if (ids.Count == 0)
            return [];

        var tasks = ids.Select(id => _externalApiService.GetByIdAsync(id, ct));
        return await Task.WhenAll(tasks);
    }
}

Допустим, RPS задаётся некими опциями, а вместе с лимитером хочется настроить некоторую политику повторов. Тогда конфигурация будет примерно такая:

services.AddResiliencePipeline<string, DataObject>(
    nameof(DataObjectServiceRateLimiterDecorator),
    (builder, pollyContext) =>
    {
        var allowedRps = pollyContext.ServiceProvider
            .GetRequiredService<IOptions<IDataObjectApiOptions>>()
            .Value.RequestsPerSecond;
        builder
            .ConfigureTelemetry(NullLoggerFactory.Instance)
            .AddRetry(
                new RetryStrategyOptions<DataObject>
                {
                    Delay = TimeSpan.FromSeconds(1),
                    MaxRetryAttempts = 5
                })
            .AddRateLimiter(
                new FixedWindowRateLimiter(
                    new FixedWindowRateLimiterOptions
                    {
                        Window = TimeSpan.FromSeconds(1),
                        PermitLimit = allowedRps,
                        QueueLimit = allowedRps / 3 + 10,
                    }));
    });

В конце концов, можно и unit-тесты написать, проверив поведение пайплайна в некоторых ситуациях. Например, вот такие:

public class DataObjectServiceRateLimiterDecoratorTests
{
    [Theory]
    [InlineData(1)]
    [InlineData(2)]
    [InlineData(3)]
    [InlineData(4)]
    [InlineData(5)]
    public async Task GetByIdAsync_ApiReturnedError_RetryCallHappened(int retryCount)
    {
        // arrange
        var response = new DataObject(Content: Guid.NewGuid().ToString());
        var apiService = new Mock<IDataObjectExternalApiService>();
        var sequentialResult = apiService.SetupSequence(
            x => x.GetByIdAsync(
                It.IsAny<int>(),
                It.IsAny<CancellationToken>()));
        for (var i = 0; i < retryCount - 1; i++)
            sequentialResult = sequentialResult.ThrowsAsync(new Exception());
        sequentialResult.ReturnsAsync(response);

        var pipelineProvider = new Mock<ResiliencePipelineProvider<string>>();
        pipelineProvider
            .Setup(
                x => x.GetPipeline<DataObject>(
                    nameof(DataObjectServiceRateLimiterDecorator)))
            .Returns(
                new ResiliencePipelineBuilder<DataObject>()
                    .AddRetry(
                        new RetryStrategyOptions<DataObject>
                        {
                            MaxRetryAttempts = retryCount,
                            Delay = TimeSpan.FromMilliseconds(1)
                        })
                    .Build());

        var decorator = new DataObjectServiceRateLimiterDecorator(
            apiService.Object, pipelineProvider.Object);

        // act
        var dataObject = await decorator.GetByIdAsync(id: default, ct: default);

        // assert
        dataObject.Should().BeEquivalentTo(response);
        apiService
            .Verify(
                x => x.GetByIdAsync(
                    It.IsAny<int>(),
                    It.IsAny<CancellationToken>()),
                Times.Exactly(retryCount));
    }

    [Theory]
    [InlineData(5, 1)]
    [InlineData(100, 50)]
    [InlineData(60, 60)]
    [InlineData(10, 11)]
    [InlineData(20, 40)]
    [InlineData(30, 100)]
    public async Task GetByIdAsync_IfRpsRateLimitExceeded_ThenExceptionIsThrown(int rps, int amount)
    {
        // arrange
        var apiService = new Mock<IDataObjectExternalApiService>();
        apiService
            .Setup(
                x => x.GetByIdAsync(
                    It.IsAny<int>(),
                    It.IsAny<CancellationToken>()))
            .ReturnsAsync(new DataObject(Content: string.Empty));

        var pipelineProvider = new Mock<ResiliencePipelineProvider<string>>();
        pipelineProvider
            .Setup(
                x => x.GetPipeline<DataObject>(
                    nameof(DataObjectServiceRateLimiterDecorator)))
            .Returns(
                new ResiliencePipelineBuilder<DataObject>()
                    .AddRateLimiter(
                        new FixedWindowRateLimiter(
                            new FixedWindowRateLimiterOptions
                            {
                                PermitLimit = rps,
                                Window = TimeSpan.FromSeconds(1)
                            }))
                    .Build());

        var decorator = new DataObjectServiceRateLimiterDecorator(
            apiService.Object, pipelineProvider.Object);

        // act
        var tasks = Enumerable.Range(0, amount)
            .Select(id => decorator.GetByIdAsync(id, ct: default));
        var ex = await Record.ExceptionAsync(() => Task.WhenAll(tasks));

        // assert
        if (amount > rps)
            ex.Should().BeOfType<RateLimiterRejectedException>();
        else
            ex.Should().BeNull();
    }

    [Fact]
    public async Task GetByIdAsync_HappyPath()
    {
        // arrange
        var apiService = new Mock<IDataObjectExternalApiService>();
        apiService
            .Setup(
                x => x.GetByIdAsync(
                    It.IsAny<int>(),
                    It.IsAny<CancellationToken>()))
            .ReturnsAsync(new DataObject(Content: Guid.NewGuid().ToString()));

        var pipelineProvider = new Mock<ResiliencePipelineProvider<string>>();
        pipelineProvider
            .Setup(
                x => x.GetPipeline<DataObject>(
                    nameof(DataObjectServiceRateLimiterDecorator)))
            .Returns(
                new ResiliencePipelineBuilder<DataObject>()
                    .AddRetry(
                        new RetryStrategyOptions<DataObject>
                        {
                            MaxRetryAttempts = 5,
                            Delay = TimeSpan.FromSeconds(1)
                        })
                    .AddRateLimiter(
                        new FixedWindowRateLimiter(
                            new FixedWindowRateLimiterOptions
                            {
                                PermitLimit = 10,
                                Window = TimeSpan.FromSeconds(1)
                            }))
                    .Build());

        var decorator = new DataObjectServiceRateLimiterDecorator(
            apiService.Object, pipelineProvider.Object);

        // act
        var tasks = Enumerable.Range(0, 100)
            .Select(id => decorator.GetByIdAsync(id, ct: default));
        var result = await Task.WhenAll(tasks);

        // assert
        result.Length.Should().Be(100);
    }
}

Заключение

Сегодня вы узнали, как потреблять внешнее API с ограничением по RPS несколькими способами. Итеративно было выявлено, что удобнее и гибче всего работать через Polly. Был предоставлен подробный пример.

Ещё я веду Telegram-канал StepOne, куда выкладываю много интересного контента про коммерческую разработку на C#, даю карьерные советы, рассказываю истории из личного опыта и раскрываю все тайны IT-индустрии.

Telegram-канал со скидками, розыгрышами призов и новостями IT 💻