Очередь задач в Python: как делать правильно

Пользователь регистрируется, ваш обработчик отправляет приветственное письмо через внешний SMTP — и ответ висит 3-4 секунды, пока идёт соединение. А если SMTP отвалился, вся регистрация падает с 500-й, хотя аккаунт уже создан. Решение — вынести медленную и ненадёжную работу из цикла запрос-ответ в фоновую очередь. Ниже — как это делают в Python, чем Celery отличается от RQ, где очередь начинает терять задачи и по какому критерию выбирать инструмент.

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

Что вообще считать очередью задач

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

Что обычно выносят в очередь:

• отправку писем, пушей, SMS — всё, что зависит от внешнего сервиса;
• тяжёлые вычисления: миниатюры, отчёты, экспорт в Excel;
• обращения к сторонним API, которые могут тормозить или отвечать 429;
• периодические задачи по расписанию — очистка, дайджесты, синхронизация.

Наивный подход и почему он ломается

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

# Так делать не нужно для продакшена
import threading

def register(request):
    user = create_user(request)
    # Поток умрёт вместе с процессом веб-сервера
    threading.Thread(target=send_welcome_email, args=(user.email,)).start()
    return ok_response()

Проблема в том, что поток живёт внутри веб-процесса. Перезапустили сервер при деплое — незавершённые задачи потеряны без следа. Упал воркер gunicorn по таймауту — задача исчезла. Нет повторов при ошибке, нет видимости, сколько задач в работе, нельзя выполнить задачу на отдельной машине. Для одноразового скрипта сойдёт, для сервиса — нет.

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

Три рабочих подхода

Celery — стандарт для сложных сценариев

Celery — самый функциональный вариант. Умеет расписания (Celery Beat), цепочки и группы задач (canvas), маршрутизацию по очередям, ограничение частоты, повторы с экспоненциальной задержкой. Работает с Redis и RabbitMQ как брокерами.

# Celery 5.x
# tasks.py
from celery import Celery

app = Celery("myproj", broker="redis://localhost:6379/0")

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_welcome_email(self, email):
    try:
        smtp_send(email, "Добро пожаловать")
    except SMTPException as exc:
        # Повтор через 60 секунд, максимум 3 раза
        raise self.retry(exc=exc)

Вызов из обработчика — send_welcome_email.delay(user.email). Воркер запускается командой celery -A tasks worker. Цена мощности — конфигурация: у Celery десятки настроек, и часть значений по умолчанию небезопасна (об этом ниже).

RQ — когда нужна простота

RQ (Redis Queue) заметно проще. Работает только с Redis, задача — обычная функция без декораторов, весь API помещается в голове.

# RQ 1.x / 2.x
from redis import Redis
from rq import Queue
from myapp.mailer import send_welcome_email

queue = Queue(connection=Redis())
# Кладём задачу; retry задаётся объектом Retry
from rq import Retry
queue.enqueue(send_welcome_email, "user@example.com",
              retry=Retry(max=3, interval=60))

Воркер поднимается командой rq worker. За простоту платите функциональностью: расписания — только через отдельный пакет rq-scheduler, сложных цепочек нет, привязка к Redis жёсткая.

Встроенные средства фреймворка

Если задача одна-две и без строгих гарантий, отдельный брокер может быть избыточным. В Django 4.2+ есть StreamingHttpResponse и обработчики, но фоновых очередей в ядре нет — их добавляют пакетами вроде django-tasks (кандидат на включение в ядро, статус уточняйте в документации версии). FastAPI даёт BackgroundTasks, но с той же оговоркой: задачи живут внутри процесса и теряются при его смерти, гарантий доставки нет.

Сравнение

Где очередь теряет задачи

Главная иллюзия — что задача, попавшая в очередь, гарантированно выполнится. На деле есть несколько мест, где всё ломается.

Очередь даёт надёжность не сама по себе, а только при правильных настройках подтверждения и хранения. Значения по умолчанию оптимизированы под скорость, а не под сохранность.

• Потеря при падении воркера. По умолчанию Celery подтверждает получение задачи в момент забора, а не после выполнения. Воркер упал в середине — задача потеряна. Лечится task_acks_late=True, но тогда задача должна быть идемпотентной, потому что при падении она выполнится повторно.
• Дубли. С поздним подтверждением одна задача может выполниться дважды. Если она списывает деньги или шлёт письмо — нужна защита от повторного выполнения на стороне самой задачи.
• Транзакция и коммит. Классическая ошибка в Django: вы кладёте задачу delay() внутри транзакции, воркер хватает её раньше, чем транзакция закоммитилась, и не находит объект в базе. Ставьте постановку задачи в transaction.on_commit().
• Redis без персистентности. Redis по умолчанию хранит данные в памяти. Перезагрузка сервера без настроенного AOF/RDB — и вся очередь исчезла. Для критичных задач это аргумент в пользу RabbitMQ или как минимум включённой персистентности Redis.
• Разросшийся бэкенд результатов. Если включён backend результатов и вы не задали TTL, результаты копятся в Redis, пока он не упрётся в память.

Когда что выбирать

• Berите RQ, если у вас уже есть Redis, задачи независимые, расписаний мало и вы хотите разобраться за час. Типичный случай — сервис среднего размера без сложных пайплайнов.
• Берите Celery, если нужны цепочки, расписания, маршрутизация по нескольким очередям с разными приоритетами или брокер надёжнее Redis. Готовьтесь потратить время на конфигурацию acks и повторов.
• Обходитесь встроенными средствами только когда потеря задачи при перезапуске некритична: например, необязательная аналитика или прогрев кеша, который всё равно построится по запросу.

Что настроить сразу после запуска

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

1. Сделайте задачи идемпотентными — они выполнятся повторно рано или поздно.
2. Настройте позднее подтверждение (task_acks_late в Celery) и осознайте, что это включает возможность дублей.
3. Задайте TTL для результатов, чтобы бэкенд не разрастался.
4. Ставьте enqueue/delay внутри on_commit, если работаете с базой.
5. Поднимите мониторинг: Flower для Celery или rq info и rq-dashboard для RQ — вам нужно видеть длину очереди и число упавших задач.
6. Настройте отдельную очередь для повторных и отложенных задач, чтобы медленные задачи не блокировали быстрые.

Дальше смотрите на два показателя в проде: длину очереди (растёт — воркеров не хватает) и количество задач в dead-letter или failed. Инвалидация зависших задач и настройка автоскейлинга воркеров — отдельная тема, здесь мы её не разворачиваем.