Планировщик задач в Python: как убрать очередь из тысячи pending

Распределённая система из 12 воркеров Celery обрабатывала поток задач по парсингу и записи в PostgreSQL. При 2 000 задач в минуту очередь в Redis начала расти: спустя 20 минут в состоянии pending висело больше 8 000 сообщений, а часть воркеров при этом простаивала с загрузкой CPU около 15 %. Классическая картина: работа есть, руки свободны, но задачи до рук не доходят. Ниже — что именно ломается в наивной конфигурации планировщика и какие изменения дали снижение времени в очереди с 4–6 минут до 20–30 секунд на том же железе.

Речь про Celery 5.3 с брокером Redis 7 и результатом в PostgreSQL 16. Замеры снимались на 12 воркерах (по 4 vCPU), поток генерировался нагрузочным скриптом на 2 000 задач/мин со смешанной длительностью: 60 % задач по 50–100 мс, 40 % по 3–5 секунд.

Где ломается наивный планировщик

Проблема почти всегда не в брокере, а в том, как воркер забирает задачи. Три механизма, которые в дефолтной конфигурации Celery работают против вас под неоднородной нагрузкой.

1. Жадный prefetch

По умолчанию воркер забирает в локальный буфер сразу несколько сообщений — это управляется настройкой worker_prefetch_multiplier (по умолчанию 4). При конкурентности 8 один воркер резервирует до 32 задач вперёд. Если среди них попались длинные, короткие задачи из того же буфера ждут своей очереди за длинными, хотя соседний воркер уже освободился. Формально задачи распределены, фактически — залипли в буфере занятого воркера.

2. Одна очередь на всё

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

3. Поштучная запись в БД

Каждая задача открывала транзакцию и делала один INSERT. При 2 000 задач/мин это 2 000 round-trip к PostgreSQL, и пул соединений становился узким местом раньше, чем CPU воркеров.

Что меняем и что это даёт

Шаг 1. prefetch_multiplier = 1

Первое и самое дешёвое изменение. Воркер забирает ровно столько задач, сколько может выполнять прямо сейчас, и не резервирует лишнее. Короткая задача больше не ждёт за длинной в чужом буфере.

# celery_app.py — Celery 5.3
from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379/0",
             backend="redis://localhost:6379/1")

app.conf.update(
    # воркер берёт по одной задаче за раз, не резервирует буфер
    worker_prefetch_multiplier=1,
    # подтверждение после выполнения, а не после получения:
    # если воркер упадёт, задача вернётся в очередь
    task_acks_late=True,
)

Осторожно с task_acks_late=True: задача должна быть идемпотентной. Если воркер упал после записи в БД, но до подтверждения, задача выполнится повторно. Для нашего парсинга спасал уникальный ключ и ON CONFLICT DO NOTHING.

Шаг 2. Разделяем очереди по длительности

Быстрые и медленные задачи разводим по разным очередям и маршрутизируем явно. Часть воркеров слушает только быструю очередь — короткие задачи перестают ждать вовсе.

app.conf.task_routes = {
    "tasks.parse_fast": {"queue": "fast"},
    "tasks.parse_slow": {"queue": "slow"},
}

# запуск воркера только на быстрой очереди:
# celery -A celery_app worker -Q fast -c 8 --prefetch-multiplier 1
# и отдельный пул на медленной:
# celery -A celery_app worker -Q slow -c 4 --prefetch-multiplier 1

Шаг 3. Батчинг записи в БД

Вместо одного INSERT на задачу собираем результаты пачками и пишем одним запросом. Используем execute_values из psycopg2 — это не отдельная транзакция на строку, а один round-trip на пачку.

# writer.py — psycopg2 2.9, PostgreSQL 16
from psycopg2.extras import execute_values

def flush_batch(connection, rows):
    """Пишет пачку строк одним запросом. rows — список кортежей."""
    if not rows:
        return
    with connection.cursor() as cursor:
        execute_values(
            cursor,
            "INSERT INTO parsed_items (url, payload) VALUES %s "
            "ON CONFLICT (url) DO NOTHING",
            rows,
            page_size=500,  # строк на один INSERT
        )
    connection.commit()

Пачку набирали до 500 строк или по таймауту 2 секунды — что наступит раньше, чтобы результаты не залёживались при спаде потока.

Цифры до и после

Замеры на одном и том же железе (12 воркеров по 4 vCPU, Redis 7, PostgreSQL 16), поток 2 000 задач/мин, окно наблюдения 20 минут.

Основной выигрыш по времени в очереди дало разделение очередей и prefetch_multiplier=1. Батчинг снял нагрузку с БД и убрал вторичное узкое место — без него при росте потока упёрлись бы в пул соединений.

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

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

Ни одно из изменений не бесплатно, и на других профилях нагрузки они дают обратный эффект.

• prefetch_multiplier=1 на очень коротких задачах. Если все задачи по 5–10 мс, отказ от буфера добавляет сетевой round-trip к брокеру на каждую задачу, и пропускная способность падает. Здесь буфер как раз нужен — держите multiplier повыше и не разводите очереди.
• Разделение очередей требует, чтобы вы заранее знали длительность. Если задача может оказаться и быстрой, и медленной в зависимости от данных, маршрутизация по имени не сработает. Тогда смотрят в сторону приоритетов внутри очереди или отдельного диспетчера.
• Батчинг увеличивает задержку одной записи. Результат появляется в БД не сразу, а по набору пачки или таймауту. Для потоковой аналитики это норма, для операций, где нужен мгновенный ответ пользователю, — нет.
• task_acks_late без идемпотентности приводит к дублям при падении воркера. Обязательно уникальный ключ и ON CONFLICT либо явная дедупликация.

Как это мерить у себя

Прежде чем крутить настройки, снимите базовые цифры — иначе не поймёте, помогло или нет.

1. Длина очереди во времени: LLEN по ключам очередей в Redis раз в 10 секунд, или через Flower / Prometheus-экспортер Celery.
2. Время задачи в очереди: разница между постановкой и стартом. В Celery ловится через сигналы task_prerun и метку времени постановки.
3. Загрузка CPU воркеров и число активных соединений к PostgreSQL (pg_stat_activity) — чтобы увидеть, куда сместилось узкое место.

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

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

Начните с worker_prefetch_multiplier=1 и разделения очередей по длительности — это дёшево и обратимо. Снимите длину очереди и время в ней до и после, на одном профиле нагрузки. Если после этого упрётесь в БД — добавляйте батчинг. Дальше стоит посмотреть на раздел Optimizing в документации Celery и на автомасштабирование пула воркеров под фактическую длину очереди, а не под фиксированное число.