Celery beat в Django: рабочие приёмы, чтобы задачи не дублировались и не терялись

Периодическая задача на Celery beat пишется за пять минут: декоратор, расписание, готово. Проблемы начинаются позже. После второго деплоя в кластере оказывается два beat-процесса, и задача уходит воркерам дважды. Ночная выгрузка молча падает, а вы узнаёте об этом от клиента. Задача, рассчитанная на 30 секунд, зависает на 20 минут и блокирует следующий запуск. Разбираем приёмы, которые снимают эти грабли на боевом проекте.

Всё ниже — для Django 4.2–5.x, Celery 5.3+ и брокера Redis либо RabbitMQ. Если храните расписание в базе — понадобится django-celery-beat 2.5+. Угол статьи узкий: только периодические задачи через beat. Ретраи обычных задач, маршрутизация по очередям и мониторинг Flower — смежные темы, здесь не разворачиваем.

Один beat на весь кластер, а не по одному на под

Beat — это отдельный процесс-планировщик. Он не выполняет задачи, а по расписанию кладёт их в брокер. Воркеры разбирают. Ключевое правило: beat должен быть ровно в одном экземпляре. Два beat = каждая задача отправляется дважды.

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

Воркеров можно и нужно много. Beat — всегда один. Это разные роли одного процесса Celery, и разносить их по контейнерам стоит с самого начала, а не когда прилетят тройные письма.

Разделяйте роли на уровне запуска. Воркеры и планировщик — отдельные команды и отдельные деплойменты:

# worker — можно реплицировать
celery -A myproject worker -l info --concurrency=4

# beat — строго одна реплика
celery -A myproject beat -l info

В Kubernetes это два отдельных Deployment: у воркера replicas любое, у beat — жёстко replicas: 1 и стратегия Recreate, чтобы во время обновления не пересеклись старый и новый под.

Расписание в коде или в базе

Есть два способа задать расписание, и выбор влияет на то, кто и как его меняет.

Если расписание фиксированное и меняется редко — держите его в коде. Динамику (пользователь сам настраивает время рассылки) отдавайте django-celery-beat с DatabaseScheduler:

# settings.py
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

# запуск beat с указанием планировщика
# celery -A myproject beat -l info -S django_celery_beat.schedulers:DatabaseScheduler

Идемпотентность: задача обязана переживать повторный запуск

Даже с одним beat задача может уйти воркеру дважды. Брокер гарантирует доставку at-least-once, а не exactly-once: при переподключении воркера или таймауте подтверждения сообщение вернётся в очередь. Поэтому задача, которая списывает деньги или шлёт письмо, должна быть безопасна к повторному выполнению.

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

from celery import shared_task
from django.core.cache import cache  # backend Redis

@shared_task
def sync_orders():
    lock_id = "lock:sync_orders"
    # cache.add вернёт False, если ключ уже существует
    got_lock = cache.add(lock_id, "1", timeout=60 * 10)
    if not got_lock:
        return "already running, skip"
    try:
        _do_sync_orders()
    finally:
        cache.delete(lock_id)

Важно: timeout на блокировке обязателен. Если воркер упадёт до finally, ключ протухнет сам, иначе задача заблокируется навсегда. Ставьте timeout заведомо больше максимального времени выполнения.

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

Ограничивайте время выполнения

Задача, зависшая на внешнем HTTP-запросе без таймаута, держит воркер и может пропустить следующий по расписанию запуск. Ставьте лимиты явно:

@shared_task(
    soft_time_limit=300,   # SoftTimeLimitExceeded — можно поймать и прибраться
    time_limit=360,        # жёсткое убийство процесса
)
def generate_report():
    from celery.exceptions import SoftTimeLimitExceeded
    try:
        _build_report()
    except SoftTimeLimitExceeded:
        # успеваем закрыть файлы, снять блокировку
        _cleanup()
        raise

Разница принципиальная. soft_time_limit бросает исключение, которое вы ловите и корректно завершаетесь. time_limit убивает процесс воркера жёстко, без шанса на уборку — держите его выше soft на запас.

Проблема overlap: запуск наложился на предыдущий

Beat запускает задачу по расписанию независимо от того, закончилась ли предыдущая. Раз в минуту стартует выгрузка, а она идёт три минуты — к третьему запуску работают три копии. У Celery нет встроенного «не запускать, пока идёт предыдущая». Решается той же Redis-блокировкой из раздела выше: наложившийся запуск не получит lock и выйдет с skip.

Часовые пояса и пропущенные запуски

Расписание Celery считает во времени, заданном настройкой CELERY_TIMEZONE. Если она не совпадает с TIME_ZONE Django, задача «раз в полночь» сработает не тогда, когда ждёте. Задавайте пояс явно и держите его согласованным:

# settings.py
CELERY_TIMEZONE = "Europe/Moscow"
CELERY_ENABLE_UTC = True

Отдельный момент — что происходит, пока beat лежал (деплой, падение). По умолчанию пропущенные за это время срабатывания не навёрстываются: crontab-расписание сработает в следующий положенный момент. Если задача критична и её нельзя пропускать, не полагайтесь на beat как на гарантию — добавьте проверку «когда последний раз отрабатывало» внутри самой задачи или сверяйте по данным.

Если не сработало: типичные симптомы

• Задача выполняется несколько раз подряд. Проверьте число beat-процессов: ps aux | grep beat на всех узлах. Почти всегда причина — второй планировщик, поднявшийся вместе с репликой.
• Beat пишет расписание в лог, но воркер молчит. Beat только отправляет в брокер. Убедитесь, что воркер подписан на ту же очередь и что имя задачи в расписании совпадает с зарегистрированным (@shared_task(name=...) либо путь через autodiscover).
• Задача не находится: NotRegistered. Модуль с задачей не импортируется при старте. Проверьте app.autodiscover_tasks() в celery.py и что файл называется tasks.py внутри приложения.
• DatabaseScheduler не видит новую задачу. После правки в админке beat перечитывает расписание не мгновенно. Он опрашивает базу с интервалом; подождите или перезапустите beat.
• Всё встало намертво после падения воркера. Скорее всего, задача взяла Redis-блокировку без timeout и ключ остался висеть. Всегда ставьте timeout на lock.

Что настроить дальше

Базовый бой держит три вещи: один beat, идемпотентность каждой периодической задачи и лимиты времени. После этого смотрите на наблюдаемость. Включите отправку событий (worker --events) и поднимите Flower или экспортируйте метрики в Prometheus — вам нужны длина очереди, время выполнения задач и число ошибок. Настройте алерт на растущую очередь: это первый признак, что воркеров не хватает либо задача зависла. И заведите отдельную задачу-«хартбит», которая просто пишет метку времени — по ней видно, что beat вообще жив.