Как реализовать систему очередей для ограничения RPS?

Ограничение частоты запросов (RPS — Requests Per Second) — это критически важный механизм для защиты сервисов от перегрузки, обеспечения справедливого распределения ресурсов и предотвращения атак типа "отказ в обслуживании" (DoS). Система очередей здесь выступает буфером: вместо немедленной обработки каждый входящий запрос помещается в очередь, а обработчик забирает задачи из неё с фиксированной, ограниченной скоростью.

Основные подходы к реализации

• Алгоритм "Дырявое ведро" (Leaky Bucket): Запросы поступают в очередь ("ведро") с любой скоростью, но обрабатываются ("вытекают") с постоянной, заданной скоростью. Это сглаживает пики нагрузки.
• Алгоритм "Ведро с токенами" (Token Bucket): В воображаемое ведро периодически добавляются "токены" (например, каждые 1/RPS секунд). Для обработки запроса необходимо извлечь токен из ведра. Если токены закончились, запрос отклоняется или ставится в очередь ожидания.
• Очередь с фиксированной задержкой: Самый простой способ — использовать очередь (например, в памяти или Redis) и отдельный процесс (воркер), который извлекает и обрабатывает задачи, искусственно добавляя задержку между операциями для соблюдения лимита RPS.

Практический пример на Node.js

Рассмотрим реализацию простого ограничителя с очередью на основе массива и таймера.

class RateLimiter {
  constructor(rps) {
    this.queue = [];
    this.rps = rps; // Максимальное количество запросов в секунду
    this.processing = false;
  }

  // Метод для добавления задачи в очередь
  addTask(task) {
    this.queue.push(task);
    this._processQueue();
  }

  // Внутренний метод для обработки очереди с заданной скоростью
  async _processQueue() {
    if (this.processing) return;
    this.processing = true;

    while (this.queue.length > 0) {
      const task = this.queue.shift(); // Извлекаем задачу из начала очереди
      try {
        await task(); // Выполняем задачу
      } catch (err) {
        console.error('Task failed:', err);
      }
      // Ждём необходимое время перед следующей задачей
      await this._delay(1000 / this.rps);
    }
    this.processing = false;
  }

  // Вспомогательная функция задержки
  _delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// Использование
const limiter = new RateLimiter(5); // Не более 5 запросов в секунду
for (let i = 0; i < 10; i++) {
  limiter.addTask(() => console.log(`Задача ${i} выполнена`));
}

В этом примере задачи добавляются в массив queue, а метод _processQueue последовательно их обрабатывает, выдерживая паузу между задачами. Для продакшена такая реализация в памяти не подойдёт — потребуется устойчивая очередь (например, на Redis) и учёт распределённой системы.

Где применяется

Ограничение RPS с очередями применяется в API-шлюзах, микросервисах, при интеграции с внешними API, имеющими жёсткие лимиты, и в любых системах, где необходимо контролировать потребление ресурсов (база данных, CPU, сеть). Готовые решения часто встроены в инструменты вроде Nginx (модуль limit_req), API-менеджеров (Kong, Apigee) или предоставляются библиотеками (например, express-rate-limit для Node.js или ratelimiter для Python).

Вывод: Систему очередей для ограничения RPS стоит реализовывать, когда ваш сервис должен гарантированно соблюдать лимиты по нагрузке, защищать себя от всплесков трафика или корректно работать с внешними ресурсами, имеющими ограничения на частоту обращений. Для простых случаев достаточно middleware в вашем фреймворке, для высоконагруженных распределённых систем — специализированных очередей сообщений (Kafka, RabbitMQ) с настроенными воркерами.