Как потреблять 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 ?