Celery в Django: фоновые и периодические задачи

Связка Django и Celery нужна, когда работу не стоит выполнять внутри HTTP-запроса: отправка писем, генерация отчётов, обработка загруженных файлов, обращения к медленным внешним API. Django ставит задачу в очередь (брокер, обычно Redis), а отдельный процесс — воркер Celery — забирает её и выполняет. Для запуска по расписанию служит планировщик celery beat.

Быстрый ответ: установите celery[redis], создайте config/celery.py с app.autodiscover_tasks(), опишите задачи через @shared_task в tasks.py приложений, вызывайте их методом .delay() и запустите воркер командой celery -A config worker -l INFO. Ниже — каждый шаг с проверенным кодом (Celery 5.6, Django 5.2 LTS и 6.x). В примерах пакет настроек проекта называется config, приложение — shop.

Установка Celery и Redis

python -m pip install "celery[redis]"

Extra [redis] подтягивает клиент redis для Python. Сам сервер Redis ставится отдельно. Проще всего запустить его в Docker:

docker run -d --name redis -p 6379:6379 redis:7

На Ubuntu подойдёт пакет redis-server (sudo apt install redis-server). Официальной сборки Redis для Windows нет — запускайте его в Docker Desktop или WSL.

Подключение Celery к проекту Django

Файл celery.py

Создайте config/celery.py рядом с settings.py:

# config/celery.py
import os

from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")

app = Celery("config")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

• DJANGO_SETTINGS_MODULE задаётся до создания приложения, чтобы воркер знал, какие настройки загрузить.
• namespace="CELERY" означает, что все настройки Celery в settings.py пишутся с префиксом CELERY_: broker_url превращается в CELERY_BROKER_URL.
• autodiscover_tasks() находит модули tasks.py во всех приложениях из INSTALLED_APPS.

Чтобы приложение Celery загружалось вместе с Django (и @shared_task знал, к какому приложению привязаться), импортируйте его в config/__init__.py:

# config/__init__.py
from .celery import app as celery_app

__all__ = ("celery_app",)

Настройки брокера Redis в settings.py

# config/settings.py
CELERY_BROKER_URL = os.environ.get("CELERY_BROKER_URL", "redis://localhost:6379/0")
CELERY_RESULT_BACKEND = os.environ.get("CELERY_RESULT_BACKEND", "redis://localhost:6379/1")
CELERY_TIMEZONE = TIME_ZONE
CELERY_TASK_TIME_LIMIT = 5 * 60  # жёсткий лимит выполнения задачи, секунд

CELERY_RESULT_BACKEND нужен, только если вы хотите получать результаты задач. Номер после слеша — номер базы Redis; брокер и результаты удобно держать в разных базах.

Задачи: @shared_task, delay и apply_async

Как объявить задачу

# shop/tasks.py
from celery import shared_task
from django.core.mail import send_mail

from .models import Order


@shared_task
def add(x, y):
    return x + y


@shared_task
def send_order_confirmation(order_id):
    order = Order.objects.select_related("customer").get(pk=order_id)
    send_mail(
        subject=f"Заказ №{order.pk} принят",
        message="Спасибо за заказ!",
        from_email=None,  # возьмётся DEFAULT_FROM_EMAIL
        recipient_list=[order.customer.email],
    )

@shared_task не привязан к конкретному объекту app, поэтому задачи переиспользуемых приложений не импортируют config.celery напрямую. Имя задачи по умолчанию — путь к функции: shop.tasks.send_order_confirmation.

Как вызвать задачу из view

from shop.tasks import add, send_order_confirmation

send_order_confirmation.delay(order.pk)

# то же самое, но с параметрами выполнения
send_order_confirmation.apply_async(args=[order.pk], countdown=60)

result = add.delay(2, 3)
result.id                  # идентификатор задачи
result.get(timeout=10)     # → 5, ждёт результат (нужен CELERY_RESULT_BACKEND)

delay(*args, **kwargs) — короткая запись apply_async(args, kwargs). Через apply_async передают опции: countdown (задержка в секундах), eta (точное время запуска), queue (имя очереди), expires.

Три правила, которые экономят часы отладки:

1. Передавайте ID, а не объекты моделей. Аргументы сериализуются в JSON; объект модели не сериализуется, а к моменту выполнения данные в базе могут измениться.
2. Ставьте задачу после коммита транзакции. Иначе воркер может начать работу раньше, чем запись появится в базе, и получить DoesNotExist. Используйте transaction.on_commit(lambda: send_order_confirmation.delay(order.pk)); в Celery 5.4+ есть сокращение send_order_confirmation.delay_on_commit(order.pk).
3. Не вызывайте result.get() внутри view. Это блокирует воркер Gunicorn и сводит пользу Celery на нет.

Запуск воркера Celery

Из каталога, где лежит manage.py, с активированным виртуальным окружением:

celery -A config worker -l INFO

-A config указывает на пакет, в котором Celery найдёт celery_app. В выводе при старте есть раздел [tasks] со списком найденных задач — если вашей задачи там нет, autodiscover_tasks() её не увидел.

Полезные флаги воркера:

Флаг	Что делает
-l, --loglevel	Уровень логов: DEBUG, INFO, WARNING, ERROR
-c, --concurrency	Число параллельных процессов (по умолчанию — число ядер)
-P, --pool	Реализация пула: prefork, solo, threads, gevent, …
-Q, --queues	Какие очереди слушать
--max-tasks-per-child	Перезапуск процесса пула после N задач

После изменения кода задач воркер нужно перезапустить: в отличие от runserver, он не перезагружается сам.

Celery на Windows: —pool=solo

Celery начиная с версии 4 официально не поддерживает Windows. Воркер с пулом по умолчанию (prefork) на Windows запускается, но задачи зависают или падают с ошибками распаковки. Для разработки используйте пул solo:

celery -A config worker -l INFO --pool=solo

solo выполняет задачи по одной в основном процессе. Если нужна параллельность, попробуйте --pool=threads. Флаг -B (встроенный beat) на Windows не работает: воркер завершится с ошибкой -B option does not work on Windows, поэтому beat запускают отдельной командой. В продакшне Celery запускают на Linux.

Повторные попытки (retry)

Задачи, которые ходят в сеть, должны переживать временные сбои. Ручной вариант — через bind=True и self.retry():

import requests
from celery import shared_task


@shared_task(bind=True, max_retries=3, default_retry_delay=10)
def notify_crm(self, order_id):
    try:
        response = requests.post(
            "https://crm.example.ru/api/orders/", json={"id": order_id}, timeout=5
        )
        response.raise_for_status()
    except requests.RequestException as exc:
        raise self.retry(exc=exc)

Автоматический вариант — параметры декоратора:

@shared_task(
    autoretry_for=(requests.RequestException,),
    retry_backoff=True,      # экспоненциальная пауза: 1, 2, 4, 8… секунд
    retry_backoff_max=600,   # но не дольше 10 минут
    retry_jitter=True,       # случайный разброс, чтобы не бить сервис залпом
    max_retries=5,
)
def fetch_exchange_rates():
    ...

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

Периодические задачи: celery beat

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

Расписание в settings.py

# config/settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "fetch-rates-every-hour": {
        "task": "shop.tasks.fetch_exchange_rates",
        "schedule": crontab(minute=0),  # в начале каждого часа
    },
    "add-every-30-seconds": {
        "task": "shop.tasks.add",
        "schedule": 30.0,
        "args": (2, 3),
    },
}

Запуск — отдельно от воркера:

celery -A config beat -l INFO

Время в crontab считается в часовом поясе CELERY_TIMEZONE (по умолчанию UTC), поэтому выше он приравнен к TIME_ZONE проекта. Для разработки на Linux и macOS можно запустить воркер и beat одним процессом: celery -A config worker -B -l INFO. В продакшне так не делают: beat должен работать ровно в одном экземпляре, иначе задачи будут дублироваться.

django-celery-beat: расписание в базе и админке

Пакет django-celery-beat хранит расписание в базе данных, и его можно менять через админку Django без перезапуска и деплоя.

python -m pip install django-celery-beat

# config/settings.py
INSTALLED_APPS = [
    # ...
    "django_celery_beat",
]

python manage.py migrate
celery -A config beat -l INFO --scheduler django_celery_beat.schedulers:DatabaseScheduler

В админке появятся разделы с интервалами, crontab-расписаниями и периодическими задачами. Создать задачу можно и из кода:

import json

from django_celery_beat.models import IntervalSchedule, PeriodicTask

schedule, _ = IntervalSchedule.objects.get_or_create(
    every=10, period=IntervalSchedule.MINUTES
)
PeriodicTask.objects.create(
    interval=schedule,
    name="Обновление курсов валют",  # имя должно быть уникальным
    task="shop.tasks.fetch_exchange_rates",
    kwargs=json.dumps({}),
)

Аргументы args и kwargs у PeriodicTask хранятся как JSON-строки, отсюда json.dumps.

Celery в продакшне

На сервере воркер и beat работают как отдельные службы systemd — так же, как Gunicorn. Минимальный unit для воркера:

# /etc/systemd/system/celery-mysite.service
[Unit]
Description=Celery worker for mysite
After=network.target redis-server.service

[Service]
User=deploy
WorkingDirectory=/srv/mysite
EnvironmentFile=/srv/mysite/.env
ExecStart=/srv/mysite/.venv/bin/celery -A config worker -l INFO
Restart=on-failure

[Install]
WantedBy=multi-user.target

Для beat создайте второй unit с командой celery -A config beat -l INFO. Общая схема сервера с Nginx, Gunicorn и systemd разобрана в статье «Деплой Django: Gunicorn и Nginx». Не забывайте перезапускать службы Celery при каждом обновлении кода.

Частые ошибки

Received unregistered task of type …

Воркер не знает задачу. Проверьте, что приложение есть в INSTALLED_APPS, задачи лежат в tasks.py, в config/__init__.py импортирован celery_app, и что воркер перезапущен после изменения кода.

delay() зависает или падает с ошибкой подключения к Redis

Celery по умолчанию пытается переподключиться к брокеру, поэтому без Redis вызов delay() может надолго зависнуть, а затем завершиться ошибкой подключения к localhost:6379. Причина — не запущен Redis или неверен CELERY_BROKER_URL. Проверьте сервер командой redis-cli ping — ответ должен быть PONG.

Задача выполнилась, но в базе старые данные

Задача ушла в очередь до коммита транзакции. Используйте transaction.on_commit() или delay_on_commit().

Object of type Order is not JSON serializable

В задачу передан объект модели. Передавайте order.pk и загружайте объект внутри задачи.

Периодическая задача запускается не в то время

Расписание crontab считается в CELERY_TIMEZONE, а не в TIME_ZONE Django — задайте его явно.

Частые вопросы

Нужен ли Celery, если в Django 6.0 появились фоновые задачи?

В Django 6.0 появился встроенный фреймворк задач django.tasks: декоратор @task и метод .enqueue(). Django берёт на себя описание задач и постановку в очередь, а выполнять их должен внешний воркер. Встроенные бэкенды ImmediateBackend и DummyBackend рассчитаны на разработку и тесты, для продакшна нужен сторонний бэкенд с воркером. Celery остаётся готовым решением с воркерами, ретраями, очередями и расписанием beat.

Можно ли использовать RabbitMQ вместо Redis?

Да, RabbitMQ — полноценный брокер Celery. Код задач не меняется, меняется только адрес брокера:

CELERY_BROKER_URL = "amqp://user:password@localhost:5672//"

Хранить результаты в RabbitMQ неудобно, поэтому CELERY_RESULT_BACKEND часто оставляют на Redis или базе данных.

Как посмотреть, какие задачи сейчас выполняются?

Команды inspect опрашивают запущенные воркеры:

celery -A config inspect active      # выполняются сейчас
celery -A config inspect scheduled   # отложены через countdown или eta

Для наблюдения в браузере есть веб-интерфейс Flower: pip install flower, затем celery -A config flower — по умолчанию он открывается на порту 5555.

Как узнать статус задачи по её ID?

Сохраните result.id после вызова delay(), а статус получите через AsyncResult. Для этого нужен CELERY_RESULT_BACKEND, иначе статус не сохраняется.

from celery.result import AsyncResult

result = AsyncResult(task_id)
result.state     # PENDING, STARTED, SUCCESS, FAILURE, RETRY
result.revoke()  # отменить задачу, если она ещё не начала выполняться

Статус STARTED появляется, только если включить CELERY_TASK_TRACK_STARTED = True.

Как тестировать задачи без воркера и Redis?

Вызывайте задачу как обычную функцию — send_order_confirmation(order.pk) выполнит её синхронно в тесте. Если нужно проверить код, который вызывает .delay(), включите в тестовых настройках CELERY_TASK_ALWAYS_EAGER = True: задачи будут выполняться сразу в текущем процессе, без брокера.

Итог

Celery подключается к Django тремя файлами: config/celery.py, импорт в config/__init__.py и tasks.py в приложениях. Задачи вызываются через delay() или apply_async(), выполняются воркером, а по расписанию их ставит celery beat — из CELERY_BEAT_SCHEDULE или из базы через django-celery-beat. Дальше — деплой Django с Gunicorn и Nginx, остальные материалы раздела собраны на странице «Деплой и инфраструктура». Если массовые действия в админке работают медленно, посмотрите, как вынести их в фон, в статье об оптимизации массовых действий.