Celery setup, tasks, beats, monitoring, error handling, retry logic
Celery — распределённая система очередей задач. Вместо того чтобы выполнять долгую работу внутри HTTP-запроса и заставлять пользователя ждать, вы кладёте задачу в очередь и отвечаете сразу. Дальше её подхватывает отдельный процесс-воркер и выполняет в фоне.
💡 Правило: всё, что занимает больше ~200 мс и не нужно пользователю прямо сейчас в ответе, выносите в Celery: отправка email, генерация PDF/отчётов, обращения к медленным внешним API, обработка загруженных файлов, рассылки.
За словом «Celery» скрывается не одна программа, а несколько ролей:
Django (producer) → Broker (очередь) → Worker (executor)
│
▼
Result Backend (опц.)
| Компонент | Роль | Чем обычно реализуют |
|---|---|---|
| Producer | Кладёт задачу в очередь (.delay()) | Ваш Django-код |
| Broker | Хранит очередь сообщений | Redis или RabbitMQ |
| Worker | Забирает и выполняет задачи | Процесс celery worker |
| Result backend | Хранит результаты/статусы задач | Redis, БД (опционально) |
| Beat | Планировщик периодических задач | Процесс celery beat |
⚠️ Redis vs RabbitMQ как брокер: Redis проще в настройке и достаточен для большинства проектов. RabbitMQ надёжнее при больших объёмах и даёт гарантии доставки из коробки. Result backend держать в Redis — нормально; хранить результаты в основной БД — антипаттерн (раздувает таблицы).
# myproject/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_TASK_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TIMEZONE = 'Europe/Moscow'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60 # жёсткий лимит — задачу убьют
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60 # мягкий — бросит исключение, можно прибраться# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
# Берём все настройки с префиксом CELERY_ из settings
app.config_from_object('django.conf:settings', namespace='CELERY')
# Автоматически находит tasks.py во всех приложениях
app.autodiscover_tasks()# myproject/__init__.py — чтобы Celery поднимался вместе с Django
from .celery import app as celery_app
__all__ = ('celery_app',)Запуск воркера и планировщика — это отдельные процессы, их не запускает runserver:
celery -A myproject worker -l info # воркер
celery -A myproject beat -l info # планировщик периодических задач
celery -A myproject worker -B -l info # оба сразу (только для dev!)# blog/tasks.py
from celery import shared_task
from django.core.mail import send_mail
@shared_task
def send_welcome_email(user_id):
from django.contrib.auth import get_user_model
user = get_user_model().objects.get(pk=user_id)
send_mail('Добро пожаловать', '...', 'noreply@site.ru', [user.email])
return user.email⚠️ Передавайте в задачу только примитивы (id, строки), а не объекты модели. Объект сериализуется в JSON и к моменту выполнения может устареть. Правильно — передать
user_idи заново достать свежий объект из БД внутри задачи.
# Поставить в очередь, вернёт сразу (не ждёт выполнения)
send_welcome_email.delay(user.id)
# Расширенный вызов с параметрами доставки
send_welcome_email.apply_async(
args=[user.id],
countdown=300, # выполнить через 5 минут
queue='emails', # положить в конкретную очередь
priority=5,
)В production особенно выручает автоматический повтор упавших задач. Внешний API «моргнул» — задача не теряется, а повторяется:
@shared_task(
bind=True, # даёт доступ к self (контексту задачи)
autoretry_for=(ConnectionError, TimeoutError),
retry_backoff=True, # экспоненциальная задержка: 1, 2, 4, 8…
retry_backoff_max=600,
retry_jitter=True, # случайный разброс, чтобы не было «грозы повторов»
max_retries=5,
)
def sync_with_crm(self, order_id):
order = Order.objects.get(pk=order_id)
response = crm_client.push(order.to_dict()) # может бросить ConnectionError
return response.status_code💡 Идемпотентность обязательна. Задача может выполниться более одного раза (повтор, сбой воркера после успеха, но до подтверждения). Пишите задачи так, чтобы повторный запуск не создавал дубль: проверяйте «уже отправлено?», используйте
update_or_create, уникальные ключи.
Beat — это «cron внутри Celery». Расписание задаётся декларативно:
# settings.py
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
'daily-report': {
'task': 'analytics.tasks.send_daily_report',
'schedule': crontab(hour=8, minute=0), # каждый день в 08:00
},
'cleanup-sessions': {
'task': 'core.tasks.clear_expired_sessions',
'schedule': crontab(minute=0, hour='*/6'), # каждые 6 часов
},
'heartbeat': {
'task': 'core.tasks.ping',
'schedule': 30.0, # каждые 30 секунд
},
}💡 Для расписаний, которые меняют не-программисты (через админку), ставьте
django-celery-beat— он хранит расписание в БД, и его можно править без деплоя.
Фоновые задачи невидимы, пока за ними не следишь. Минимум для production:
celery -A myproject flower. Показывает активные/упавшие задачи, воркеров, очереди.# Проверить результат конкретной задачи
result = send_welcome_email.delay(user.id)
result.id # task_id для трекинга
result.status # PENDING / STARTED / SUCCESS / FAILURE / RETRY
result.ready() # завершилась ли| Ошибка | Последствие | Решение |
|---|---|---|
| Передача ORM-объекта вместо id | Устаревшие данные, ошибки сериализации | Передавать pk, доставать в задаче |
| Задача не идемпотентна | Дубли писем/платежей при повторе | Проверки уникальности, флаги |
Нет time_limit | Зависшая задача держит воркер вечно | task_time_limit, soft_time_limit |
| Тяжёлый результат в backend | Раздувание Redis/БД | ignore_result=True, если результат не нужен |
| Одна очередь на всё | Долгие задачи блокируют быстрые | Разные очереди + выделенные воркеры |
# Результат не нужен — не засоряем backend
@shared_task(ignore_result=True)
def write_audit_log(event): ...# imports/tasks.py
from celery import shared_task
from django.core.mail import mail_admins
@shared_task(bind=True, soft_time_limit=600, max_retries=3)
def process_csv_import(self, import_job_id):
job = ImportJob.objects.get(pk=import_job_id)
job.status = 'processing'
job.save(update_fields=['status'])
try:
created = 0
for row in read_csv(job.file.path):
Product.objects.update_or_create( # идемпотентно
sku=row['sku'], defaults={'price': row['price']},
)
created += 1
job.status = 'done'
job.rows_processed = created
except SoftTimeLimitExceeded:
job.status = 'timeout'
raise
except Exception as exc:
job.status = 'failed'
mail_admins('Импорт упал', str(exc))
raise self.retry(exc=exc, countdown=60)
finally:
job.save()# views.py — кладём задачу и сразу отвечаем
def upload_import(request):
job = ImportJob.objects.create(file=request.FILES['file'])
process_csv_import.delay(job.id)
return JsonResponse({'job_id': job.id, 'status': 'queued'})autodiscover_tasks()@shared_task, вызываете через .delay()/.apply_async()id, а не объекты моделиtime_limit/soft_time_limitВопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.
Далее: Docker и Деплой