Celery в Python: настройка фоновых задач, повторов, расписания и разделения очередей

Тяжёлые вычисления, построение аналитических отчётов, отправка email и вызовы внешних API блокируют основной поток веб-приложения, если выполнять их синхронно в контексте HTTP-запроса. Для переноса таких операций в фон в экосистеме Python стандартом является распределённая очередь задач Celery.

Ниже подробно рассмотрены архитектура Celery, запуск фоновых задач на примере генерации отчётов, мониторинг их выполнения, механизмы автоматических повторов (retries), настройка планировщика Celery Beat и разделение воркеров по типам нагрузок.

---

Архитектура Celery: брокер, воркер и Result Backend

Celery работает по принципу передачи сообщений между продюсером (клиентским приложением на Django, FastAPI, Flask) и потребителями (воркерами).

Как взаимодействуют компоненты системы

1. Producer (Клиент): веб-сервис принимает запрос пользователя, упаковывает параметры задачи в сообщение и отправляет его в брокер.
2. Broker (Брокер сообщений): промежуточное хранилище (чаще всего Redis или RabbitMQ), которое принимает задачи в очереди и удерживает их до момента обработки.
3. Worker (Воркер): независимый процесс (или пул процессов), который непрерывно опрашивает брокер, забирает задачи и исполняет связанный с ними Python-код.
4. Result Backend (Хранилище результатов): база данных (Redis, PostgreSQL, Memcached), в которую воркер записывает финальный статус выполнения (успех, ошибка, повтор) и возвращаемое значение функции.

Разница между брокером сообщений и бэкендом результатов

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

Result Backend подключается отдельно, если вызывающему сервису необходимо проверять статус задачи (PENDING, STARTED, SUCCESS, FAILURE) или забирать возвращённый результат. Если результат задачи сохраняется напрямую в основную базу данных или S3-хранилище, Result Backend можно отключить для экономии сетевых и вычислительных ресурсов.

---

Сквозной пример: фоновое формирование отчёта

Рассмотрим практический сценарий: пользователь запрашивает выгрузку финансового отчёта за год. Генерация занимает продолжительное время.

Инициализация приложения Celery и регистрация задачи

Создадим файл tasks.py:

import time
from celery import Celery

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

# Настройка формата сериализации и таймзоны
app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Europe/Moscow",
    enable_utc=True,
)

@app.task
def generate_pdf_report(report_id: int, user_email: str) -> dict:
    # Имитация ресурсоёмкой выборки данных и компиляции PDF
    time.sleep(5)
    file_path = f"/storage/reports/report_{report_id}.pdf"
    
    return {
        "report_id": report_id,
        "file_path": file_path,
        "user_email": user_email,
        "status": "ready"
    }

Постановка задачи в очередь: delay() против apply_async()

Отправить задачу в очередь можно двумя методами:

# 1. Быстрый вызов через delay()
# Принимает аргументы самой функции
result = generate_pdf_report.delay(42, "user@example.com")

# 2. Расширенный вызов через apply_async()
# Позволяет управлять параметрами отправки, очередями и задержками
result = generate_pdf_report.apply_async(
    args=[42, "user@example.com"],
    queue="reports",
    countdown=10,  # Запустить задачу не ранее чем через 10 секунд
    expires=3600   # Время жизни задачи в очереди (в секундах)
)

Метод delay() является сокращённой обёрткой над apply_async(). Для базовых вызовов достаточно delay(), но при необходимости маршрутизации или отложенного старта используется apply_async().

---

Контроль состояния задачи и получение результата

Оба метода — delay() и apply_async() — возвращают объект AsyncResult. С его помощью можно опрашивать состояние задачи из любого места приложения, зная её UUID (task_id).

Жизненный цикл задачи и стандартные статусы

• PENDING — задача ожидает в очереди или её task_id неизвестен бэкенду.
• STARTED — воркер начал выполнение (требует опции task_track_started = True).
• RETRY — произошёл сбой, задача отправлена на повтор.
• SUCCESS — функция успешно выполнена, результат сохранён.
• FAILURE — выполнение завершилось необработанным исключением.

Проверка прогресса с помощью AsyncResult

from celery.result import AsyncResult
from tasks import app

def check_task_status(task_id: str):
    res = AsyncResult(task_id, app=app)
    
    response = {
        "task_id": task_id,
        "status": res.status,
        "ready": res.ready(),
        "successful": res.successful()
    }
    
    if res.ready():
        if res.successful():
            response["result"] = res.result
        else:
            response["error"] = str(res.result)
            response["traceback"] = res.traceback
            
    return response

В HTTP-обработчиках веб-сервера не рекомендуется вызывать блокирующий метод res.get(), так как он замораживает поток веб-приложения до окончания расчёта, нивелируя преимущества асинхронной очереди.

---

Обработка сбоев и повторные попытки (Retry)

При обращении к сторонним базам данных или сетевым API возможны временные сбои. Celery предоставляет механизмы перезапуска упавших задач.

Ручной перезапуск через self.retry() и параметр countdown

Чтобы управлять логикой повтора внутри тела функции, используется флаг bind=True. Он передаёт экземпляр текущей задачи в качестве первого аргумента self:

import requests
from celery import Celery

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

class NetworkFetchError(Exception):
    pass

@app.task(bind=True, max_retries=3)
def fetch_external_data(self, source_url: str):
    try:
        response = requests.get(source_url, timeout=5)
        response.raise_for_status()
        return response.json()
    except (requests.RequestException, NetworkFetchError) as exc:
        # Повторить задачу через 30 секунд
        raise self.retry(exc=exc, countdown=30)

Автоматические повторы с autoretry_for и экспоненциальной задержкой (retry_backoff)

Для декларативного описания повторов применяется параметр autoretry_for. Это исключает громоздкие блоки try/except:

@app.task(
    autoretry_for=(requests.RequestException,),
    retry_kwargs={"max_retries": 5},
    retry_backoff=True,          # Экспоненциальное увеличение задержки (1s, 2s, 4s, 8s...)
    retry_backoff_max=300,       # Максимальная пауза между повторами (5 минут)
    retry_jitter=True            # Добавление случайного интервала для предотвращения Thundering Herd
)
def fetch_api_metrics(endpoint: str):
    response = requests.get(endpoint, timeout=3)
    return response.json()

Требование идемпотентности при ретраях

Повторный запуск задачи не должен приводить к повреждению данных или дублированию операций.

• Неидемпотентная операция: списание баланса пользователя или отправка письма клиенту без проверки статуса в базе данных.
• Идемпотентная операция: перезапись файла отчёта по фиксированному пути, обновление статуса записи с проверкой текущего состояния или вызов внешнего API с передачей уникального ключа идемпотентности (Idempotency-Key).

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

---

Разделение коротких и длительных задач по очередям

Смешивание мгновенных задач (отправка SMS, генерация кодов подтверждения) и долгих операций (построение отчётов, конвертация видео) в одном пуле воркеров приводит к блокировке очередей.

Почему нельзя смешивать быстрые и тяжёлые процессы в одном пуле

Если пул состоит из 4 процессов, а в очередь поступило 4 отчёта длительностью по 2 минуты каждый, воркер будет полностью занят. Срочная задача на отправку письма с авторизацией встанет в хвост очереди и будет ожидать освобождения процессов, ухудшая отклик системы.

Маршрутизация задач и запуск воркеров с флагом -Q

Для изоляции потоков настраивается маршрутизация по разным очередям:

# Настройка маршрутизации в конфигурации Celery
app.conf.task_routes = {
    "tasks.send_instant_notification": {"queue": "high_priority"},
    "tasks.generate_pdf_report": {"queue": "heavy_reports"},
}

Запуск независимых воркеров в терминале или через systemd / Docker Compose:

# 1. Воркер для быстрых критичных задач (выделяем больше параллельных процессов)
celery -A tasks worker --loglevel=INFO -Q high_priority -c 8 -n worker_fast@%h

# 2. Воркер для тяжёлых вычислений (ограничиваем конкурентность для экономии CPU/RAM)
celery -A tasks worker --loglevel=INFO -Q heavy_reports -c 2 -n worker_heavy@%h

Флаг -c (--concurrency) задаёт число дочерних процессов или потоков. Тяжёлые задачи больше не замедляют выполнение мгновенных уведомлений.

---

Периодические задачи с Celery Beat

Если задачу требуется запускать по расписанию (например, формировать сводный отчёт каждую ночь в 04:00), используется планировщик Celery Beat.

Принцип работы планировщика Beat

Celery Beat — это отдельный служебный процесс. Он не исполняет саму логику задач, а считывает расписание и вовремя отправляет сообщения в брокер. Непосредственное выполнение производят стандартные воркеры.

[Celery Beat] ---> (По расписанию кладет задачу в Redis) ---> [Worker]

Настройка расписания beat_schedule

from celery.schedules import crontab
from tasks import app

app.conf.beat_schedule = {
    # Запуск каждые 15 минут
    "check-system-health-every-15-mins": {
        "task": "tasks.check_health",
        "schedule": 900.0,
        "options": {"queue": "high_priority"}
    },
    # Запуск отчёта каждый понедельник в 04:00 утра
    "weekly-financial-report-monday": {
        "task": "tasks.generate_pdf_report",
        "schedule": crontab(hour=4, minute=0, day_of_week="monday"),
        "args": (101, "admin@example.com"),
        "options": {"queue": "heavy_reports"}
    },
}

Запуск планировщика:

celery -A tasks beat --loglevel=INFO

---

Рекомендации по оптимизации и частые ошибки

1. Не передавайте сложные ORM-объекты в аргументы:
   • Не рекомендуется: generate_report.delay(user_model_instance) — объект базы данных может устареть к моменту выполнения задачи или вызвать ошибку сериализации.
   • Рекомендуется: generate_report.delay(user.id) — передавайте примитивные типы (int, str), а сущность запрашивайте из БД уже внутри задачи.
2. Не храните бинарные файлы в Result Backend:
   • Result Backend чувствителен к объёму памяти. Сохраняйте готовые файлы в объектное хранилище (S3/MinIO), а в результат задачи возвращайте путь или URL.
3. Учитывайте предвыборку (worker_prefetch_multiplier):
   • По умолчанию воркер заранее резервирует несколько задач из брокера. Для долгих операций рекомендуется установить worker_prefetch_multiplier = 1, чтобы один воркер не удерживал пачку тяжёлых задач, пока остальные простаивают.
4. Настраивайте подтверждения (task_acks_late):
   • По умолчанию задача подтверждается брокеру сразу при получении. Если воркер аварийно завершится в процессе работы, задача будет утеряна. Для критических расчётов включают task_acks_late = True, чтобы подтверждение отправлялось строго после успешного завершения функции.

---

Вопросы и ответы (FAQ)

1. В чём разница между .delay() и .apply_async()?

delay() — это сокращённый синтаксис для запуска задачи с параметрами по умолчанию. apply_async() даёт полный контроль над вызовом: позволяет задать очередь (queue), задержку запуска (countdown), точное время выполнения (eta), ограничение по времени (time_limit) и параметры повторов.

2. Нужно ли сохранять тело сформированного файла в Result Backend?

Нет. Result Backend предназначен для статусов и компактных метаданных. Запись больших бинарных данных (PDF, архивы, изображения) в Redis или PostgreSQL приводит к перерасходу оперативной памяти и снижению производительности. Файлы сохраняют в файловую систему или S3, возвращая из задачи идентификатор или ссылку.

3. Что произойдёт, если процесс воркера будет принудительно завершён во время формирования отчёта?

Если включен режим по умолчанию (task_acks_late = False), задача считается подтверждённой в момент взятия в работу и будет потеряна. Если установлен параметр task_acks_late = True, брокер зафиксирует разрыв соединения с воркером и перенаправит задачу другому доступному процессу.

4. Почему Celery Beat и Celery Worker запускаются как два разных процесса?

Beat — это планировщик событий, отслеживающий интервалы времени, а Worker — вычислительный процесс. Разделение изолирует расписание от возможных сбоев или зависаний вычислительного пула.

5. Как защититься от зависания воркера при обработке внешних данных?

Используйте параметры time_limit (принудительное завершение процесса по SIGKILL) и soft_time_limit (генерация исключения SoftTimeLimitExceeded, позволяющего корректно закрыть дескрипторы и освободить ресурсы перед остановкой).

---

Заключение

Celery предоставляет надёжную инфраструктуру для организации асинхронных и фоновых вычислений в Python. Разделение очередей по типу нагрузки предотвращает задержки критичных операций, экспоненциальные повторы с autoretry_for обеспечивают устойчивость к сетевым сбоям, а планировщик Beat автоматизирует периодическую обработку данных. При проектировании фоновых задач важно соблюдать идемпотентность, передавать в аргументы простые типы данных и изолировать длительные вычисления в выделенные пулы воркеров.

Источники

• Configuration and defaults — Celery 5.6.3 documentation
• Tasks — Celery 5.6.3 documentation - docs.celeryq.dev
• Next Steps — Celery 5.6.3 documentation
• Concurrency — Celery 5.6.3 documentation
• Optimizing — Celery 5.6.3 documentation
• celery.app.autoretry — Celery 5.7.0a1 documentation