Django Celery и Valkey: брокер, beat и лимиты памяти

Celery в Django - это три части обвязки и четыре настройки, которые не стоит оставлять по умолчанию. Обвязка: файл celery.py рядом с settings.py, импорт в __init__.py и настройки с префиксом CELERY_, которые читаются из обычного модуля настроек Django. Настройки: acks_late, множитель prefetch, visibility timeout и потолок памяти на дочерний процесс. Брокером служит Valkey, и так как он говорит на протоколе Redis, Celery подключается к нему по обычному URL вида redis://.

На маленьком тарифе проблемы возникают не из-за конфигурации, а из-за арифметики. Воркер Celery - это копия вашего Django-проекта: одна для родительского процесса и ещё по одной на каждый дочерний, и рядом стоит gunicorn, то есть тот же проект ещё раз. В этой статье мы настроим брокер и расписание beat, а большую часть текста посвятим тому, сколько процессов помещается в купленную память и что делать, если не помещается. Предполагается, что проект уже развёрнут так, как описано в деплое Django в production с gunicorn, и сравнение библиотек, которое есть в статье очереди задач на Valkey, здесь не повторяется.

Что вы строите#

Три вида процессов используют один Valkey. Веб-процесс (gunicorn) ставит задачу в очередь и возвращает ответ. Один или несколько воркеров забирают задачи из очереди и выполняют их. Один процесс beat, и ровно один, по часам добавляет в очередь периодические задачи. Сообщения между ними хранит Valkey.

Две вещи на этой схеме определяют всё дальнейшее. Beat никогда не выполняет задачи, он только отправляет их, поэтому он крошечный. А воркеру нужны собственные подключения к базе данных, потому что он импортирует ваши модели и работает с базой точно так же, как веб-процесс. Как они складываются, посчитано в статье пулы соединений и лимиты.

Установка Celery для Django#

Установите Celery с дополнением для транспорта Redis, которое подтягивает клиентскую библиотеку, работающую и с Valkey:

$ pip install "celery[redis]"

Зафиксируйте версию в requirements.txt, как и всё остальное. Имена ключей конфигурации Celery изменились в серии 4.x, а некоторые значения по умолчанию поменялись между релизами 5.x, поэтому диапазон версий, плавающий через минорный релиз, - это способ получить изменившееся поведение при перезапуске в три часа ночи. О файлах блокировки зависимостей рассказано в статье requirements и virtualenv в Python.

Интеграция с Django состоит из трёх файлов. Сначала приложение Celery, в том же пакете, где лежит settings.py:

import osfrom celery import Celeryos.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings.production")app = Celery("myproject")app.config_from_object("django.conf:settings", namespace="CELERY")app.autodiscover_tasks()

Затем убедитесь, что оно импортируется при старте Django, чтобы декораторы @shared_task привязывались именно к этому приложению:

from .celery import app as celery_app__all__ = ("celery_app",)

namespace="CELERY" означает, что каждая настройка Celery в settings.py имеет префикс CELERY_ и пишется в верхнем регистре: настройка Celery task_acks_late превращается в CELERY_TASK_ACKS_LATE. autodiscover_tasks() импортирует модуль tasks.py из каждого приложения в INSTALLED_APPS. Задача, которая лежит где-то ещё, не регистрируется, и когда придёт сообщение для неё, воркер напишет в лог Received unregistered task.

Задачи кладите в tasks.py внутри приложения и используйте shared_task, чтобы приложение не импортировало экземпляр Celery:

from celery import shared_taskfrom .models import Order@shared_task(    autoretry_for=(ConnectionError, TimeoutError),    retry_backoff=True,    retry_backoff_max=600,    retry_jitter=True,    max_retries=5,)def send_receipt(order_id):    order = Order.objects.get(pk=order_id)    order.send_receipt_email()

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

Настройки брокера#

Настройки подключения и поведения лежат в модуле настроек Django и читаются из окружения, чтобы пароля не было в репозитории:

import osCELERY_BROKER_URL = os.environ["VALKEY_URL"]          # redis://:password@host:port/0CELERY_RESULT_BACKEND = os.environ["VALKEY_URL"].replace("/0", "/1")CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = TrueCELERY_TASK_ACKS_LATE = TrueCELERY_TASK_REJECT_ON_WORKER_LOST = TrueCELERY_WORKER_PREFETCH_MULTIPLIER = 1CELERY_TASK_TIME_LIMIT = 300CELERY_TASK_SOFT_TIME_LIMIT = 240CELERY_RESULT_EXPIRES = 3600CELERY_TASK_IGNORE_RESULT = TrueCELERY_BROKER_TRANSPORT_OPTIONS = {    "visibility_timeout": 3600,    "global_keyprefix": "shop:",}CELERY_TIMEZONE = "UTC"

У каждой из них есть своя причина.

URL берётся из данных подключения вашего тарифа Valkey: это хост, порт и сгенерированный пароль, и ничто из этого не совпадает со значениями по умолчанию локальной установки. Если в пароле есть такие символы, как @, / или #, закодируйте их в URL процентной записью (@ становится %40), иначе клиент разрежет строку не в том месте и выдаст путаную ошибку про хост. Это делает urllib.parse.quote из Python.

CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP важна при перезапуске. Без неё воркер, стартовавший за несколько секунд до того, как брокер стал доступен, завершается, а свежие релизы Celery 5.x выводят предупреждение об устаревании с просьбой её задать. Панель хостинга после деплоя перезапускает всё сразу, так что порядок запуска вам неподконтролен.

Номер базы после косой черты - это логическая база внутри того же Valkey. Если положить результаты в базу 1, они окажутся вне пространства ключей очереди, и вы сможете смотреть или очищать одно, не трогая другое. Если вы никогда не читаете возвращаемые значения задач, самый дешёвый вариант - CELERY_TASK_IGNORE_RESULT = True и вообще без бэкенда результатов, это убирает целый класс остаточных ключей.

Acks, prefetch и visibility timeout#

В транспорте Redis нет встроенного подтверждения, поэтому Celery строит его сам. Сообщение, взятое воркером, перемещается в скрытую структуру unacked. Если воркер подтверждает его в пределах visibility timeout, сообщение удаляется; если нет, оно возвращается в очередь, и другой воркер выполняет его.

Отсюда три правила.

1. Visibility timeout должен быть больше самой долгой вашей задачи, включая обратные отсчёты её повторных попыток. По умолчанию он равен одному часу. Задача, которая выполняется девяносто минут при значении по умолчанию, доставляется второй раз, пока первая ещё работает. Это самая частая причина жалобы «моя задача выполнилась дважды» на брокере с протоколом Redis. Учтите, что задачи с ETA и countdown до наступления срока удерживает воркер, а не Valkey, поэтому задача, назначенная на завтра, при таймауте в час будет доставляться повторно каждый час. Короткие таймауты и длинные отсчёты несовместимы; работу в далёком будущем планируйте через beat.
2. При `acks_late` задача может выполниться больше одного раза. Если воркер убит посреди задачи, сообщение возвращается, и задача стартует заново с самого начала. Делайте задачи идемпотентными: уникальное ограничение на идентификатор события или столбец статуса, который проверяется первым. CELERY_TASK_REJECT_ON_WORKER_LOST распространяет это и на дочерний процесс, убитый ядром из-за памяти, а на маленьком тарифе важен именно этот случай.
3. Prefetch 1 меняет пропускную способность на справедливость. Множитель 4 по умолчанию означает, что каждый дочерний процесс заранее резервирует четыре сообщения. Если одно из них - десятиминутный отчёт, три сообщения за ним ждут, даже если другие дочерние процессы простаивают. Для задач разной длины правильно 1; для тысяч мелких одинаковых задач значение по умолчанию быстрее.

У acks_late без reject_on_worker_lost есть ещё одна ловушка: если дочерний процесс убит убийцей по нехватке памяти (OOM killer), в некоторых версиях сообщение подтверждается как неудачное, а не отправляется на повтор. Именно установка обеих настроек заставляет убийство по памяти вести себя как повтор.

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

from django.db import transactiondef create_order(request):    order = Order.objects.create(...)    transaction.on_commit(lambda: send_receipt.delay(order.pk))

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

Beat: периодическое расписание#

Beat читает расписание и отправляет задачи в назначенное время. Это отдельный процесс, и на одно расписание должен работать ровно один. Два beat означают, что каждая периодическая задача отправляется дважды. Это самый важный факт о масштабировании Celery: воркеров можно добавлять свободно, но второй beat добавлять нельзя.

Простейшее расписание лежит в настройках:

from celery.schedules import crontabCELERY_BEAT_SCHEDULE = {    "clear-expired-sessions": {        "task": "accounts.tasks.clear_sessions",        "schedule": crontab(hour=3, minute=30),    },    "daily-digest": {        "task": "notifications.tasks.send_digest",        "schedule": crontab(hour=7, minute=0, day_of_week="mon-fri"),        "options": {"expires": 3600},    },    "sync-stock": {        "task": "shop.tasks.sync_stock",        "schedule": 300.0,        "args": (),    },}

Просто число - это интервал в секундах, отсчитываемый от запуска beat. crontab срабатывает по настенным часам и использует CELERY_TIMEZONE; задайте её явно, потому что по умолчанию это UTC, и дайджест «в 07:00» иначе приходит не в 07:00 по вашему времени. Если синтаксис cron вам знаком смутно, справочник - статья выражения cron простыми словами, с оговоркой, что crontab() в Celery принимает именованные поля, а не строку из пяти частей.

Опцию expires стоит скопировать. Если воркеры не работали час, расписание, срабатывающее каждые пять минут, оставит ждать двенадцать одинаковых сообщений sync_stock, и все они выполнятся пачкой, когда воркеры вернутся. С expires устаревшие отбрасываются непрочитанными.

Запустите beat отдельным процессом:

$ celery -A myproject beat --loglevel=INFO -s /tmp/celerybeat-schedule

-s задаёт, где beat хранит celerybeat-schedule - небольшой файл, в котором записано, когда каждая запись выполнялась в последний раз. По умолчанию он пишется в рабочий каталог. Beat спокойно переживает потерю этого файла при перезапуске: он тогда вычисляет, что пора выполнять, по часам, и для записей crontab это верно, а для интервальных означает «начать отсчёт заново». Задача, которая должна выполняться ровно один раз в сутки, должна быть записью crontab, а не интервалом в 86400 секунд.

Вам встретится совет запускать beat внутри воркера через celery worker -B. Это экономит процесс и годится для одного воркера на небольшом сервере, но именно из-за этого у людей появляются два beat в тот день, когда они добавляют второй воркер. Пользуйтесь этим осознанно, с явно заданным --concurrency и числом воркеров, зафиксированным на одном.

Для расписаний, которые сотрудники должны править без деплоя, django-celery-beat хранит их в базе данных и добавляет страницы в админке:

$ pip install django-celery-beat$ celery -A myproject beat --scheduler django_celery_beat.schedulers:DatabaseScheduler

Добавьте "django_celery_beat" в INSTALLED_APPS и выполните python manage.py migrate. Платой станет ещё один набор таблиц и расписание, которое больше не лежит в системе контроля версий. Для горстки фиксированных заданий оставьте словарь выше.

Память и concurrency на маленьком тарифе#

Именно здесь Celery на маленьком сервере либо выживает, либо погибает. Пул Celery по умолчанию, prefork, запускает один родительский процесс и затем --concurrency дочерних. Concurrency по умолчанию равна числу ядер CPU, которые видит процесс, а на общем хосте это могут быть ядра всей машины, а не ваша доля. Задайте её сами.

Каждый дочерний процесс импортирует ваш Django-проект, поэтому стоит примерно столько же, сколько один воркер gunicorn: обычно 100-200 МБ для скромного проекта и больше при тяжёлых зависимостях. Как это измерить, объясняет статья сколько RAM нужно приложению на Python. Родительский процесс добавляет примерно столько же. Воркер, запущенный с --concurrency 2, - это, значит, три процесса, а вместе с gunicorn рядом проект загружен пять или шесть раз.

Это отправные точки, построенные из приведённых выше цифр по процессам, а не измерения вашего кода. Прежде чем фиксировать схему, запустите воркер под нагрузкой и посмотрите резидентный размер каждого процесса командой ps -o pid,rss,cmd -C celery. На RE:NODE консоль рисует график памяти относительно лимита тарифа, пока вы это делаете, и лимит - это выключатель: при достижении лимита памяти ядро останавливает контейнер, и он перезапускается начисто, вместо того чтобы уходить в swap. Для воркера Celery это гибель всех дочерних процессов разом, и именно поэтому acks_late и reject_on_worker_lost выше не опциональны.

Четыре меры удерживают воркер в рамках бюджета.

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

$ celery -A myproject worker --pool=threads --concurrency=8 --loglevel=INFO

Потоки делят процесс Django, поэтому утечка в одной задаче утекает для всех, а обычные оговорки о потокобезопасности относятся к вашему собственному коду. Для нагрузки вроде писем и вебхуков экономия велика: восемь одновременных задач примерно за память одной. Есть ещё --pool=solo - по одной задаче за раз в главном процессе; это наименьший возможный след и разумный выбор на тарифе в 1 GB при редких заданиях.

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

CELERY_WORKER_MAX_TASKS_PER_CHILD = 200CELERY_WORKER_MAX_MEMORY_PER_CHILD = 300_000   # килобайты, около 300 МБ

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

Уберите лишнюю болтовню. На маленьком тарифе три флага убирают трафик и процессы, которые не нужны, когда воркер один:

$ celery -A myproject worker --concurrency=2 -Ofair \    --without-gossip --without-mingle --without-heartbeat

-Ofair заставляет родительский процесс отдавать работу только свободным дочерним, что важно при длинных задачах и prefetch больше 1. --without-gossip, --without-mingle и --without-heartbeat прекращают обмен состоянием кластера с другими воркерами, что полезно, когда других нет. Без heartbeat команды celery status и inspect с других узлов менее информативны, поэтому оставьте его включённым, если запускаете несколько воркеров и полагаетесь на эти команды.

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

CELERY_TASK_DEFAULT_QUEUE = "default"CELERY_TASK_ROUTES = {    "reports.tasks.*": {"queue": "slow"},}

$ celery -A myproject worker -Q default --concurrency=2 -n fast@%h$ celery -A myproject worker -Q slow    --concurrency=1 -n slow@%h

Медленной очереди достаётся один дочерний процесс, так что тяжёлая выгрузка может занять не больше доли одного ядра и памяти одного процесса, а всё срочное продолжает идти. Цена - ещё один родительский процесс. На совсем маленьком тарифе дешевле оставить одну очередь и один воркер и смириться с тем, что срочные задачи ждут.

CPU - это отдельный лимит, не связанный с памятью. На RE:NODE показатель CPU - это жёсткое ограничение до купленной доли, поэтому воркер с --concurrency 4 на половине ядра не выполняет четыре задачи со скоростью четырёх ядер; он выполняет их медленно, но все сразу. Concurrency выше доли CPU помогает только задачам, которые проводят время в ожидании чего-то другого.

Где работают процессы#

Веб-процесс, воркер и beat - это три долгоживущие команды. У сервера приложений на RE:NODE одна команда запуска, поэтому прямолинейная схема - отдельные серверы Python из одного репозитория, у каждого своя команда запуска, и все смотрят в один и тот же Valkey и базу данных через переменные окружения. Веб-сервер запускает gunicorn, сервер воркера - celery -A myproject worker, а процесс beat добавляется к команде воркера или получает собственный небольшой сервер. Это стоит дороже одного тарифа и снимает вопрос о том, какой процесс будет убит, когда закончится память.

Если вы предпочитаете держать всё на одном сервере, сработает shell-команда запуска, которая стартует воркер в фоне, а gunicorn на переднем плане, но отдавайте себе отчёт, от чего вы отказываетесь: если воркер умрёт, его никто не перезапустит, пока gunicorn жив, и очередь перестанет разгребаться без какого-либо видимого сбоя. Смотрите в консоль после каждого деплоя и для всего важного предпочитайте отдельные серверы.

Тарифы Valkey начинаются с 256 MB за $2 в месяц, а тарифы Python - с $4. Сервер Valkey доступен на собственном хосте и порту со сгенерированным паролем, поэтому веб-серверы и серверы воркеров подключаются к нему одинаково, где бы ни работали. Есть также комплект из тарифа Python, тарифа PostgreSQL и тарифа Valkey на 256 MB, который покупается вместе со скидкой на вкладке Bundles в магазине панели, и он повторяет форму всей этой статьи: веб-приложение, база данных и брокер.

Git-деплои применяются к каждому серверу отдельно. Push в репозиторий перезапускает каждый сервер, у которого включён deploy-on-push, так что воркеры получают новый код тем же push, что и веб-процесс. Работающий воркер Celery держит старый код в памяти до перезапуска, поэтому это важно. Перезапуск, пришедший посреди задачи, - это обычная остановка, и acks_late означает, что прерванная задача будет доставлена снова. Почему сигнал остановки должен давать задаче завершиться, рассказано в статье корректное завершение и health checks.

Настройки Valkey, нужные брокеру Celery#

Три серверные настройки решают, переживут ли сообщения сбой. Статья очереди задач на Valkey разбирает их для каждой библиотеки; краткая версия для Celery:

• Вытеснение. Проверьте maxmemory-policy на вашем экземпляре. Любое значение, кроме noeviction, может удалить список очереди celery или структуру unacked, когда память заполнена. Политики объясняет статья память Valkey и политики вытеснения.
• Сохранение на диск. Тариф Valkey на RE:NODE сохраняет данные на диск через AOF и снимки, так что поставленные в очередь задачи переживают перезапуск. Чего стоит каждый способ, объясняет статья AOF и RDB.
• Размер. Сообщение занимает от нескольких сотен байт до нескольких килобайт. Экземпляр на 256 MB вмещает очень большой отставший хвост мелких сообщений; заполняется он бэкендом результатов, у которого никто не задал срок жизни, или большими аргументами.

Посмотреть, что создал Celery, можно через valkey-cli:

$ valkey-cli -h HOST -p PORT --askpass> KEYS *> LLEN celery> HLEN unacked> INFO memory

LLEN celery - число ожидающих сообщений в очереди по умолчанию, а HLEN unacked - число взятых, но ещё не подтверждённых. KEYS * годится на маленьком экземпляре и является дурной привычкой на большом; там используйте SCAN. О подключении рассказано в статье начало работы с Valkey.

Команда celery -A myproject inspect active покажет, что каждый воркер выполняет прямо сейчас, celery -A myproject inspect stats - цифры по пулу и памяти, а celery -A myproject purge выбросит всё ожидающее; она просит подтверждения, и решение это нужно принимать осознанно.

Устранение неполадок#

Воркер запускается и ничего не делает. URL брокера неверен так, что ошибка возникает не сразу, или задачи лежат в другой очереди, чем та, которую слушает воркер. Проверьте celery -A myproject inspect registered, затем LLEN для имени очереди.

`Received unregistered task of type`. Воркер не импортировал модуль, в котором определена задача. Проверьте autodiscover_tasks(), что приложение есть в INSTALLED_APPS и что файл называется tasks.py.

`kombu.exceptions.OperationalError` или `Connection refused`. Хост, порт или пароль неверны, либо у веб-процесса нет сетевого пути до Valkey. Сначала проверьте тот же URL через valkey-cli с того же сервера.

Задача выполняется дважды. Visibility timeout короче задачи, или дочерний процесс был убит до подтверждения. Увеличьте таймаут, включите acks_late вместе с reject_on_worker_lost и сделайте задачу идемпотентной.

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

Периодические задачи не выполняются никогда. Beat не запущен или запущен с другим модулем настроек, чем тот, где лежит CELERY_BEAT_SCHEDULE. Когда всё работает, beat пишет в лог Sending due task для каждой из них.

Сервер перезапускается каждые несколько часов. Убийство по памяти. Сравните сумму воркеров веб-части, родительского процесса и дочерних с тарифом через ps, затем примените меры выше. Еженедельный перезапуск по расписанию на вкладке Schedules - грубый, но действенный запасной вариант; о выборе времени рассказано в статье расписания перезапуска, которые помогают.

`DatabaseError: connection already closed` внутри задачи. Долгоживущий воркер сохранил соединение, которое база данных закрыла. Вызывайте django.db.close_old_connections() в начале долгих задач или держите CONN_MAX_AGE небольшим в настройках, которые использует воркер.

Задачи копятся, а воркеры простаивают. Часто это множитель prefetch в сочетании с одной очень долгой задачей. Задайте его равным 1 и используйте -Ofair.

FAQ#

Нужен ли Celery для проекта на Django?

Часто нет. Management-команда, запускаемая по расписанию, покрывает ночные чистки и дайджесты без брокера, а очередь на базе данных справляется со скромными объёмами. Celery оправдан, когда задачи частые, требуют повторов с нарастающей паузой и должны переживать перезапуск. Варианты сравниваются в статье фоновые задачи на маленьком сервере.

Может ли Celery использовать Valkey и как брокер, и как бэкенд результатов?

Да. Используйте один и тот же сервер с другим номером базы, как в настройках выше, либо задайте CELERY_TASK_IGNORE_RESULT = True и обойдитесь без бэкенда. Альтернатива именно для Django - django-celery-results, которая хранит результаты в вашей базе данных при CELERY_RESULT_BACKEND = "django-db".

Сколько воркеров Celery запускать?

Столько дочерних процессов, чтобы хвост очереди в пик оставался около нуля, и не больше, чем позволяет память. Для задач, упирающихся в сеть, пул потоков из 8-20 в одном процессе дешевле восьми дочерних процессов prefork. Для задач, упирающихся в CPU, - один дочерний процесс на каждое реально имеющееся ядро. Правильный ответ для маленького тарифа почти всегда меньше значения по умолчанию.

Почему число ключей в Valkey продолжает расти?

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

Безопасно ли запускать beat на каждом сервере?

Нет. Ровно один процесс beat на расписание, иначе каждая периодическая задача отправляется по разу на каждый beat. Поставьте его на один сервер или встройте через -B в единственный воркер, который вы никогда не масштабируете.

Что происходит с задачами в очереди при перезапуске Valkey?

Если сохранение на диск включено, очередь загружается с диска, а воркеры переподключаются, когда задана CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP. Без сохранения все ожидающие задачи пропадают. Задачи из unacked возвращаются в очередь после visibility timeout.