Celery и фоновые задачи

Celery — распределённая очередь задач для Python: код отправляет задачу в брокер, а отдельные процессы-воркеры её выполняют.

Зачем выносить работу из HTTP-запроса:
- долгие операции — отчёты, импорт, обработка видео и картинок; пользователь не ждёт, запрос не упирается в таймаут;
- внешние вызовы — письма, SMS, push, вебхуки; падение стороннего сервиса не ломает ответ пользователю;
- повторы — задачу можно автоматически повторить при ошибке;
- периодические задачи — рассылки, очистка, синхронизация по расписанию (celery beat);
- сглаживание нагрузки — пики складываются в очередь, воркеры разбирают её в своём темпе.

@app.task
def send_welcome_email(user_id):
    user = User.objects.get(id=user_id)
    mailer.send(user.email, 'Добро пожаловать!')

# во view: ответ уходит сразу, письмо отправит воркер
send_welcome_email.delay(user.id)

- Producer (клиент) — ваше приложение (Django, FastAPI), которое вызывает task.delay(). Задача сериализуется (JSON) и отправляется в брокер.
- Broker — очередь сообщений: RabbitMQ или Redis (реже SQS). Хранит задачи, пока их не заберёт воркер.
- Worker — процесс celery worker: забирает задачи из очередей и выполняет их. Можно запустить сколько угодно воркеров на разных машинах.
- Result backend (необязательно) — куда сохраняются статус и результат задачи: Redis, БД.
- Beat (необязательно) — планировщик, который по расписанию кладёт задачи в брокер.

Django ──delay()──▶ Broker (RabbitMQ/Redis) ──▶ Worker ×N ──▶ Result backend
                         ▲
                  Celery beat (расписание)

Важно: воркер — отдельный процесс со своей копией кода. После деплоя нового кода воркеры тоже нужно перезапустить.

delay(*args, **kwargs) — короткая форма apply_async только с аргументами задачи.

apply_async принимает параметры выполнения:

send_report.delay(user_id, 'pdf')

send_report.apply_async(
    args=(user_id, 'pdf'),
    countdown=60,                 # выполнить не раньше чем через 60 с
    # eta=datetime(...),          # или в конкретное время
    expires=3600,                 # не выполнять, если не успели за час
    queue='reports',              # в какую очередь
    priority=5,
    retry=True,                   # повторять отправку в брокер при сбое соединения
)

Обе возвращают AsyncResult с id задачи.

Не путать: вызов send_report(user_id, 'pdf') без delay выполнит задачу синхронно в текущем процессе, как обычную функцию.

RabbitMQ — полноценный брокер сообщений, рекомендуемый для Celery:
- надёжные подтверждения (ack) — задача не потеряется при падении воркера;
- сложная маршрутизация (exchanges), приоритеты, TTL, dead-letter-очереди;
- persistent-очереди на диске;
- ещё один сервис в инфраструктуре, который нужно поддерживать.

Redis — проще в эксплуатации и часто уже есть в проекте (кэш):
- быстро и легко настроить;
- нет настоящих ack: используется visibility_timeout — неподтверждённая задача снова станет видимой только через это время (по умолчанию 1 час);
- задачи с eta/countdown дольше visibility_timeout могут выполниться повторно;
- при падении Redis без persistence задачи теряются.

Как выбирать: небольшой проект и Redis уже есть — Redis. Важна надёжность доставки, много очередей и сложная маршрутизация — RabbitMQ.

Result backend хранит статус и результат задачи, чтобы получить их по id:

result = generate_report.delay(42)
result.id          # 'c2f1...'
result.status      # PENDING / STARTED / SUCCESS / FAILURE / RETRY
result.get(timeout=10)

Не всегда нужен: большинство задач (отправить письмо, обновить кэш) не возвращают ничего полезного. Для них ignore_result=True — экономия памяти Redis и запросов.

Подводные камни:
- result.get() внутри HTTP-запроса или другой задачи блокирует процесс и убивает смысл асинхронности (Celery даже запрещает это внутри задач);
- статус PENDING возвращается и для несуществующего id — Celery не отличает «ещё не начата» от «не было такой задачи»;
- результаты нужно чистить (result_expires).

Частая практика: статус долгой задачи хранят в своей таблице (ReportJob), а не в result backend — так проще показывать прогресс и историю.

Автоматические повторы:

@app.task(
    autoretry_for=(requests.RequestException,),
    retry_backoff=True,          # 1, 2, 4, 8... секунд
    retry_backoff_max=600,
    retry_jitter=True,           # случайный разброс, чтобы не бить толпой
    max_retries=5,
)
def sync_order(order_id):
    crm.push(Order.objects.get(id=order_id))

Ручной повтор:

@app.task(bind=True, max_retries=3)
def charge(self, payment_id):
    try:
        gateway.charge(payment_id)
    except GatewayTimeout as exc:
        raise self.retry(exc=exc, countdown=30)

Правила:
- повторять только временные ошибки (сеть, таймаут, 5xx, 429), а не баги в коде и 4xx;
- экспоненциальная задержка с jitter;
- ограничивать число попыток и обрабатывать окончательный провал: лог, алерт, статус «failed», отдельная очередь для разбора;
- задача должна быть идемпотентной, иначе повтор задвоит эффект.

По умолчанию (acks_late=False) воркер подтверждает (ack) задачу брокеру до выполнения. Если процесс убьют посреди работы (OOM, деплой, kill -9), задача потеряна — at-most-once.

С acks_late=True подтверждение отправляется после выполнения. Если воркер умер, брокер вернёт задачу в очередь, и её выполнит другой воркер — at-least-once.

@app.task(acks_late=True, reject_on_worker_lost=True)
def process_payment(payment_id): ...

# или глобально
app.conf.task_acks_late = True
app.conf.worker_prefetch_multiplier = 1

Цена: задача может выполниться больше одного раза → она обязана быть идемпотентной.

Нюансы:
- если задача упала с исключением, она всё равно подтверждается — acks_late защищает от смерти процесса, а не от ошибок в коде;
- reject_on_worker_lost=True нужен, чтобы задача вернулась в очередь, если дочерний процесс воркера был убит;
- с Redis-брокером задача вернётся только через visibility_timeout.

Задача может выполниться дважды: повтор после таймаута, acks_late после падения воркера, повторная отправка. Идемпотентная задача при повторе не меняет результат.

Приёмы:
- Проверка состояния перед действием:

@app.task(acks_late=True)
def send_receipt(order_id):
    order = Order.objects.get(id=order_id)
    if order.receipt_sent_at:          # уже сделано — выходим
        return
    mailer.send_receipt(order)
    Order.objects.filter(id=order_id).update(receipt_sent_at=now())

- Ключ идемпотентности у внешнего вызова: платёжке передаётся idempotency_key=f'order-{order_id}', и повторный запрос не спишет деньги второй раз.
- Уникальные ограничения в БД и get_or_create / INSERT ... ON CONFLICT DO NOTHING.
- Абсолютные значения вместо приращений: status = 'paid', а не balance += 100.
- Блокировка (select_for_update или lock в Redis), если два воркера могут взять одну задачу одновременно.

Передавайте идентификаторы, а не объекты:

# плохо
send_email.delay(user)              # ORM-объект: не сериализуется в JSON, данные устареют

# хорошо
send_email.delay(user.id)           # воркер сам загрузит актуальное состояние

Почему:
- аргументы сериализуются (JSON по умолчанию) — модели, файлы, соединения так не передать;
- между постановкой задачи и выполнением могут пройти секунды или часы — объект успеет измениться;
- большие аргументы раздувают брокер. Большие данные кладут в хранилище (S3, БД), а в задачу передают ссылку.

Исключение: если нужно состояние на момент события (сумма заказа в момент оплаты), передают именно эти значения явно.

Не используйте pickle-сериализацию: она позволяет выполнить произвольный код при чтении сообщения из брокера.

Гонка:

with transaction.atomic():
    order = Order.objects.create(...)
    send_confirmation.delay(order.id)   # задача уже в брокере
    # ... ещё код, транзакция не закоммичена

Воркер может взять задачу раньше, чем транзакция закоммитится, и получить Order.DoesNotExist. А если транзакция откатится, письмо уйдёт о заказе, которого нет.

Решение — ставить задачу после коммита:

with transaction.atomic():
    order = Order.objects.create(...)
    transaction.on_commit(lambda: send_confirmation.delay(order.id))

# Celery 5.4+: то же короче
send_confirmation.delay_on_commit(order.id)

Остаётся риск: коммит прошёл, а процесс упал до отправки в брокер — задача потеряна. Если это критично, используют transactional outbox: запись о задаче сохраняется в БД в той же транзакции, а отдельный процесс отправляет её в очередь.

Зависшая задача (бесконечный цикл, внешний API без таймаута) навсегда занимает слот воркера.

@app.task(soft_time_limit=60, time_limit=90)
def build_report(report_id):
    try:
        do_heavy_work(report_id)
    except SoftTimeLimitExceeded:
        mark_report_failed(report_id)   # успеваем прибраться
        raise

- soft_time_limit — внутри задачи бросается исключение SoftTimeLimitExceeded, можно корректно завершиться.
- time_limit (hard) — процесс воркера убивается и заменяется новым. Без шанса прибраться.

Дополнительно:
- таймауты на все внешние вызовы внутри задачи (HTTP, БД);
- expires — не выполнять задачу, которая слишком долго пролежала в очереди (утренняя рассылка вечером уже не нужна);
- soft limit работает только в пуле prefork (на сигналах).

celery -A project worker --pool=prefork --concurrency=4

- prefork (по умолчанию) — несколько дочерних процессов. Обходит GIL, подходит для CPU-задач и большинства случаев. Каждый процесс потребляет память.
- threads — потоки в одном процессе. Подходит для I/O-задач; на CPU-задачах упирается в GIL.
- gevent / eventlet — зелёные потоки, сотни и тысячи одновременных I/O-задач (HTTP-запросы, вебхуки). Требуют совместимых библиотек (monkey patching).
- solo — одна задача за раз в главном процессе. Для отладки и Windows.

Concurrency:
- CPU-задачи — примерно по числу ядер;
- I/O-задачи — больше (или gevent с --concurrency=200);
- на маленьком сервере ограничивает память: каждый процесс prefork держит копию приложения.

Практика: разные типы задач — в разные очереди и на разные воркеры с подходящим пулом.

Prefetch — воркер заранее забирает из брокера несколько задач в свой буфер. Их число — worker_prefetch_multiplier × concurrency (по умолчанию 4 × concurrency).

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

Решение для длинных задач:

app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True

Воркер берёт следующую задачу только после завершения текущей.

Для массы коротких задач больший prefetch, наоборот, ускоряет работу: меньше обращений к брокеру.

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

app.conf.task_routes = {
    'notifications.tasks.*': {'queue': 'fast'},
    'reports.tasks.*': {'queue': 'heavy'},
}

# или при вызове
build_report.apply_async(args=(42,), queue='heavy')

celery -A project worker -Q fast  --concurrency=8
celery -A project worker -Q heavy --concurrency=2 --prefetch-multiplier=1

Зачем:
- срочные задачи (письмо со сбросом пароля, OTP) не ждут за тяжёлыми отчётами;
- разные ресурсы: CPU-задачам — мощные машины, I/O-задачам — gevent;
- независимое масштабирование: выросла очередь отчётов — добавляем воркеров только для неё;
- изоляция сбоев: проблемная интеграция не забивает все воркеры.

Приоритеты внутри одной очереди тоже есть (RabbitMQ — x-max-priority, Redis — эмуляция), но отдельные очереди надёжнее и нагляднее.

Причины роста памяти:
- утечки в коде или библиотеках, кэши на уровне модуля;
- задачи, которые загружают в память всю таблицу (list(Order.objects.all()));
- фрагментация памяти: Python неохотно возвращает память ОС;
- DEBUG=True в Django — сохраняет все SQL-запросы в памяти.

Что делать:

celery -A project worker --max-tasks-per-child=100 --max-memory-per-child=300000

- max_tasks_per_child — перезапускать процесс после N задач;
- max_memory_per_child (в КБ) — перезапускать процесс, превысивший лимит;
- обрабатывать данные пачками (iterator(), chunked), не держать большие объекты;
- DEBUG=False на проде;
- ограничить concurrency под доступную память, иначе придёт OOM killer.

Celery beat — отдельный процесс-планировщик: по расписанию кладёт задачи в брокер. Выполняют их обычные воркеры.

from celery.schedules import crontab

app.conf.beat_schedule = {
    'daily-digest': {
        'task': 'notifications.tasks.send_digest',
        'schedule': crontab(hour=9, minute=0),
    },
    'cleanup-every-10-min': {
        'task': 'core.tasks.cleanup',
        'schedule': 600.0,
    },
}
app.conf.timezone = 'Europe/Moscow'

Подводные камни:
- beat должен быть ровно один. Два экземпляра — каждая задача выполнится дважды. В Kubernetes это отдельный deployment с replicas: 1;
- beat лежит — задачи просто не ставятся, и ошибок нет. Нужен мониторинг;
- если очередь забита, задача выполнится позже расписания или накопятся дубли → expires и идемпотентность;
- часовой пояс и переход на летнее время;
- django-celery-beat хранит расписание в БД и позволяет менять его из админки.

apply_async(countdown=86400) выглядит как способ «выполнить через сутки», но:

- воркер забирает такую задачу сразу и держит в памяти до наступления срока. Тысячи отложенных задач — рост памяти воркера;
- с брокером Redis задача, которая не подтверждена дольше visibility_timeout (по умолчанию 1 час), снова станет видимой, и её возьмёт другой воркер — дубль;
- отменить задачу сложно: revoke хранится в памяти воркеров и теряется при перезапуске;
- после деплоя в очереди висят задачи со старыми аргументами.

Как лучше для длинных отсрочек: хранить в БД запись «что и когда сделать» (scheduled_at) и периодической задачей beat забирать созревшие:

@app.task
def send_due_reminders():
    for reminder in Reminder.objects.filter(scheduled_at__lte=now(), sent=False)[:500]:
        send_reminder.delay(reminder.id)

Такую запись легко отменить, перенести и увидеть.

Canvas — примитивы для объединения задач в процессы. Signature (task.s(...)) — задача с аргументами, которую можно передать и вызвать позже.

from celery import chain, group, chord

# последовательно: результат каждой передаётся следующей
chain(download.s(url), resize.s(), upload.s())()

# параллельно
group(resize.s(img_id) for img_id in ids)()

# параллельно, затем колбэк со списком результатов
chord(
    (count_orders.s(shop_id) for shop_id in shops),
    build_summary.s(),
)()

- .si() — immutable signature: не принимает результат предыдущей задачи.
- link_error — колбэк при ошибке.

Подводные камни:
- chord требует result backend и чувствителен к сбоям: одна упавшая задача — колбэк не выполнится;
- сложные цепочки трудно отлаживать и перезапускать с середины;
- для длинных бизнес-процессов надёжнее хранить состояние в БД (state machine) и двигать его отдельными задачами.

Что важно видеть:
- длину очередей — растёт, значит, воркеры не успевают;
- время ожидания в очереди и время выполнения задач;
- долю ошибок и повторов по типам задач;
- живы ли воркеры и beat.

Инструменты:
- Flower — веб-интерфейс: воркеры, задачи, графики, ручной revoke;
- celery inspect active / reserved / scheduled, celery status;
- метрики в Prometheus (celery-exporter) и алерты в Grafana;
- длина очереди напрямую из брокера (LLEN в Redis, management API RabbitMQ);
- ошибки задач — в Sentry (есть интеграция с Celery).

Алерты: очередь растёт дольше N минут, воркеров 0, beat не ставил задачи дольше интервала, всплеск FAILURE.

Warm shutdown (SIGTERM) — воркер перестаёт брать новые задачи и дожидается текущих. Cold shutdown (второй сигнал или SIGQUIT) — прерывает задачи немедленно.

Проблема: Docker и Kubernetes ждут после SIGTERM ограниченное время (10 и 30 секунд по умолчанию), а затем шлют SIGKILL. Долгая задача будет убита посреди выполнения.

Что делать:
- увеличить период ожидания (stop_grace_period в compose, terminationGracePeriodSeconds в k8s) под типичную длительность задач;
- acks_late=True — убитая задача вернётся в очередь;
- задачи идемпотентные и короткие: длинную работу разбивать на части;
- сигнал должен доходить до процесса Celery: запуск через exec, без промежуточного shell.

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

1. Тестировать логику как обычную функцию — без брокера и воркера:

def test_send_receipt_marks_order():
    order = OrderFactory()
    send_receipt(order.id)               # прямой вызов, синхронно
    order.refresh_from_db()
    assert order.receipt_sent_at is not None

2. Проверять, что задача поставлена, а не выполнять её:

def test_signup_schedules_email(mocker):
    delay = mocker.patch('accounts.views.send_welcome_email.delay')
    client.post('/signup', data)
    delay.assert_called_once()

3. task_always_eager=True — задачи выполняются синхронно при delay(). Удобно, но поведение отличается от реального: нет сериализации аргументов, нет отдельного процесса, ошибки проще. В Django ещё и on_commit не срабатывает в TestCase без captureOnCommitCallbacks.

4. Интеграционные тесты с реальным воркером — фикстуры celery_app и celery_worker из celery.contrib.pytest.

- Celery — самый функциональный: много брокеров, canvas, beat, маршрутизация. Цена — сложность конфигурации и неочевидное поведение по умолчанию.
- RQ — простая очередь на Redis. Минимум настроек, подходит для небольших проектов.
- Dramatiq — проще Celery и с более надёжными настройками по умолчанию (ack после выполнения, ретраи из коробки). RabbitMQ или Redis.
- arq / taskiq — асинхронные очереди под asyncio: естественны для FastAPI.
- Kafka — не очередь задач, а лог событий: хранит сообщения, позволяет перечитывать, гарантирует порядок в партиции, у каждой группы потребителей своя позиция чтения. Для событий между сервисами, а не для «выполни функцию в фоне».

Как выбирать: фоновые задачи внутри одного приложения — Celery, Dramatiq или RQ; асинхронный стек — arq или taskiq; поток событий между сервисами — Kafka.

Celery (до версии 6) не поддерживает async def задачи напрямую: воркер вызывает задачу как обычную функцию.

@app.task
def fetch_prices(ids):
    return asyncio.run(_fetch_prices(ids))

async def _fetch_prices(ids):
    async with httpx.AsyncClient(timeout=10) as client:
        responses = await asyncio.gather(*(client.get(f'/prices/{i}') for i in ids))
    return [r.json() for r in responses]

Подводные камни:
- asyncio.run создаёт новый event loop на каждую задачу — клиенты и пулы, созданные на уровне модуля для другого цикла, сломаются (attached to a different loop). Создавайте их внутри корутины;
- асинхронные соединения с БД (например, asyncpg) так же привязаны к циклу;
- с пулом gevent смешивать asyncio не стоит.

Альтернатива: если проект целиком асинхронный, удобнее взять очередь, нативно поддерживающую asyncio (arq, taskiq).

- Задача ставится внутри транзакции → воркер не находит объект. Нужен transaction.on_commit.
- ORM-объекты в аргументах вместо id.
- Неидемпотентные задачи при acks_late и ретраях → двойные письма и списания.
- Нет таймаутов на внешние вызовы и time_limit → воркеры висят.
- Одна очередь на всё → срочные задачи ждут за тяжёлыми.
- Ретраи на любые исключения, включая баги, → бесконечный шум в логах.
- Два экземпляра beat → задачи по расписанию выполняются дважды.
- Большой countdown/eta с брокером Redis → дубли из-за visibility_timeout.
- result.get() в HTTP-запросе → синхронное ожидание вместо асинхронности.
- result backend включён для всех задач → Redis забивается ненужными результатами.
- Нет мониторинга очередей → о проблеме узнают от пользователей.
- Воркеры не перезапущены после деплоя → выполняется старый код.