Обратное давление в конвейере: цифры до и после

При входном потоке 1000 событий в секунду и пропускной способности обработчиков 700 событий очередь без ограничения вырастет на 18 000 элементов за минуту. Обратное давление не ускорит обработку, но не даст продюсеру бесконечно складывать работу в память: очередь упрётся в заданный предел, а источник начнёт ждать. Ниже — разбор на Python 3.12 с расчётами, измерениями и границами применимости.

Что именно ломается без обратного давления

Типичный конвейер состоит из источника событий, очереди и нескольких обработчиков. Источник читает сообщения быстрее, чем workers успевают их обработать. Если очередь неограниченная, система продолжает принимать данные и выглядит здоровой до тех пор, пока не растёт задержка или не заканчивается RAM.

Возьмём воспроизводимую модель:

• источник производит 1000 событий в секунду;
• четыре обработчика каждый справляется со 175 событиями в секунду;
• суммарная пропускная способность — 700 событий в секунду;
• одно событие занимает в памяти 4 КБ вместе с метаданными Python-объекта и обвязкой приложения.

Разница скоростей равна 300 событий в секунду. За 60 секунд в очереди окажется 18 000 элементов. При оценке 4 КБ это около 72 МБ только под полезные записи. Реальное потребление будет выше из-за накладных расходов объектов, аллокатора и ссылок.

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

Как устроено ограниченное ожидание

В asyncio.Queue параметр maxsize задаёт верхнюю границу числа элементов. Когда очередь заполнена, await queue.put(event) приостанавливает корутину до освобождения места. Это и есть обратное давление на уровне приложения.

В нашей модели разумный предел — 500 событий. При разнице скоростей 300 событий в секунду он заполнится примерно за 1,67 секунды. После этого источник не будет продолжать бесконтрольный рост: он начнёт проводить время в ожидании свободного слота.

Сноска. Backpressure, или обратное давление, — механизм, при котором медленный этап заставляет предыдущий этап снизить скорость, временно остановиться или отклонить часть работы.

Минимальная схема для asyncio 3.12

Ниже показан фрагмент с bounded queue и четырьмя workers. Вызов task_done() должен соответствовать каждому успешному get(), иначе ожидание queue.join() не завершится.

import asyncio
import time

QUEUE_LIMIT = 500
WORKER_COUNT = 4

async def process_event(event: dict) -> None:
    # Здесь находится реальная работа: запрос к БД, HTTP или запись в файл.
    await asyncio.sleep(0.005)

async def worker(queue: asyncio.Queue) -> None:
    while True:
        event = await queue.get()
        try:
            await process_event(event)
        finally:
            queue.task_done()

async def producer(queue: asyncio.Queue, events: list[dict]) -> None:
    for event in events:
        wait_started = time.perf_counter()
        await queue.put(event)
        waited_seconds = time.perf_counter() - wait_started
        if waited_seconds > 0.01:
            print(f"producer waited {waited_seconds:.3f}s")

async def main() -> None:
    queue = asyncio.Queue(maxsize=QUEUE_LIMIT)
    workers = [asyncio.create_task(worker(queue)) for _ in range(WORKER_COUNT)]

    events = [{"id": number} for number in range(10_000)]
    await producer(queue, events)
    await queue.join()

    for task in workers:
        task.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

asyncio.run(main())

Фрагмент демонстрирует именно ограничение очереди, а не готовую интеграцию с брокером. Для Kafka, RabbitMQ или внешнего API нужно отдельно настроить подтверждения, повторную доставку и остановку чтения. Эти вопросы не сводятся к maxsize.

Цифры до и после: что меняется на самом деле

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

Ключевой результат — не рост throughput, а ограничение бэклога. Если источник обязан принять все события, ожидание нужно вынести за пределы процесса: в брокер с долговечным хранением или в другой буфер. Если данные можно потерять, после заполнения очереди допустимы отказ, дроп или выборка только последних событий. Это разные политики, и подмена одной другой создаёт уже не техническую, а бизнес-проблему.

Что измерять вместо субъективного «стало быстрее»

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

1. Глубина очереди. Снимайте текущее и максимальное значение. Постоянное приближение к 500 означает, что система работает у предела.
2. Время ожидания на put. В коде выше это интервал от начала до завершения queue.put. Нулевое ожидание говорит, что ограничение пока не срабатывает.
3. Возраст самого старого события. Очередь из 500 быстрых задач и очередь из 500 задач, ожидающих пять минут, — разные аварии.
4. Время обработки. Разделяйте p50, p95 и p99. Среднее скрывает редкие медленные обращения к БД или внешнему API.
5. Число отказов и повторов. Особенно важно, если upstream получает таймаут после заполнения очереди.

Тест проводите при фиксированных условиях: версия Python, число workers, размер события, задержка внешней зависимости, лимиты CPU и памяти. Прогоните систему минимум в трёх режимах: ниже номинальной нагрузки, на границе пропускной способности и с устойчивым превышением входного потока. В отчёте храните не только «до» и «после», но и настройки каждого прогона.

Где подход ломается

Слишком маленькая очередь

Лимит 1–10 элементов быстро передаст задержку upstream. Это может быть правильным выбором для интерактивного запроса, но плохим — для пакетной загрузки, где короткие всплески должны сглаживаться. Размер очереди выбирайте по допустимой задержке и объёму памяти, а не по круглому числу.

Слишком большая очередь

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

Workers блокируют event loop

Ограниченная asyncio.Queue не превращает синхронный вызов в асинхронный. Блокирующая библиотека для файлов, CPU или сети может остановить весь event loop. В результате очередь будет выглядеть небольшой, но фактическая пропускная способность упадёт. Для CPU-bound работы нужны отдельные процессы или иной подход к параллелизму; для блокирующего I/O — совместимый асинхронный клиент либо аккуратно вынесенная операция.

Нет политики при переполнении

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

Как выбрать размер очереди

Начните с бюджета задержки. Если вы допускаете буферизацию на 2 секунды, а обработчики справляются с 700 событий в секунду, стартовый лимит можно оценить как 1400 элементов. Это не окончательная настройка: при всплеске в 1000 событий в секунду и пропускной способности 700 очередь всё равно будет заполняться, просто медленнее.

Проверьте три ограничения:

• сколько памяти занимает один элемент вместе с обвязкой;
• какую задержку готов принять пользователь или downstream-сервис;
• что должно произойти, когда предел достигнут.

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

Что делать дальше

Сначала добавьте метрики глубины очереди, времени ожидания и возраста старого события без изменения лимита. Затем проведите нагрузочный прогон и зафиксируйте точку, где p95 задержки начинает расти. После этого задайте bounded queue, определите поведение при переполнении и проверьте корректное завершение workers. В продакшене первым делом поставьте алерт не только на заполнение очереди, но и на скорость её роста: именно она показывает, что текущая пропускная способность уже не покрывает входной поток.