Celery django что это

Как настроить Celery в Django

В этом руководстве по использованию Celery совместно с Django я расскажу:

1. Как настроить Celery с Django.
2. Как протестировать Celery-задачу в Django-оболочке.
3. Где контролировать работу Celery-приложения.

Вы можете использовать на исходный код проекта из этого репозитория.

Зачем приложению на Django нужен Celery

Celery нужен для запуска задач в отдельном рабочем процессе ( worker ), что позволяет немедленно отправить HTTP-ответ пользователю в веб-процессе (даже если задача в рабочем процессе все еще выполняется). Цикл обработки запроса не будет заблокирован, что повысит качество взаимодействия с пользователем.
Ниже приведены некоторые примеры использования Celery:

• Вы создали приложение с функцией отправки комментариев, в которых пользователь может использовать символ @, чтобы упомянуть другого пользователя, после чего последний получит уведомление по электронной почте. Если пользователь упоминает 10 человек в своем комментарии, веб-процессу необходимо обработать и отправить 10 электронных писем. Иногда это занимает много времени (сеть, сервер и другие факторы). В данном случае Celery может организовать отправку писем в фоновом режиме, что в свою очередь позволит вернуть HTTP-ответ пользователю без ожидания.
• Нужно создать миниатюру загруженного пользователем изображения? Такую задачу стоит выполнить в рабочем процессе.
• Вам необходимо делать что-то периодически, например, генерировать ежедневный отчет, очищать данные истекшей сессии. Используйте Celery для отправки задач рабочему процессу в назначенное время.

Когда вы создаете веб-приложение, постарайтесь сделать время отклика не более, чем 500мс (используйте New Relic или Scout APM), если пользователь ожидает ответа слишком долго, выясните причину и попытайтесь устранить ее. В решении такой проблемы может помочь Celery.

Celery или RQ

RQ (Redis Queue) — еще одна библиотека Python, которая решает вышеуказанные проблемы.
Логика работы RQ схожа с Celery (используется шаблон проектирования производитель/потребитель). Далее я проведу поверхностное сравнение для лучшего понимания, какой из инструментов более подходит для задачи.

• RQ (Redis Queue) проста в освоении, направлена на снижение барьера в использовании асинхронного рабочего процесса. В ней отсутствуют некоторые функции, и она работает только с Redis и Python.
• Celery предоставляет больше возможностей, поддерживает множество различных серверных конфигураций. Одним из минусов такой гибкости является более сложная документация, что довольно часто пугает новичков.

Я предпочитаю Celery, поскольку он замечательно подходит для решения многих проблем. Данная статья написана мной, чтобы помочь читателю (особенно новичку) быстро изучить Celery!

Брокер сообщений и бэкенд результатов

Брокер сообщений — это хранилище, которое играет роль транспорта между производителем и потребителем.
Из документации Celery рекомендуемым брокером является RabbitMQ, потому что он поддерживает AMQP (расширенный протокол очереди сообщений).

Так как во многих случаях нам не нужно использовать AMQP, другой диспетчер очереди, такой как Redis, также подойдет.

Бэкенд результатов — это хранилище, которое содержит информацию о результатах выполнения Celery-задач и о возникших ошибках.

Здесь рекомендуется использовать Redis.

Как настроить Celery

Celery не работает на Windows. Используйте Linux или терминал Ubuntu в Windows.

Далее я покажу вам, как импортировать Celery worker в ваш Django-проект.

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

Используйте Docker для подготовки среды разработки

Если вы работаете в Linux или Mac, у вас есть возможность использовать менеджер пакетов для настройки Redis (brew, apt-get install), однако я хотел бы порекомендовать вам попробовать применить Docker для установки сервера redis.

1. Вы можете скачать Docker-клиент здесь.
2. Затем попробуйте запустить службу Redis $ docker run -p 6379: 6379 —name some-redis -d redis

Команда выше запустит Redis на 127.0.0.1:6379.

1. Если вы намерены использовать RabbitMQ в качестве брокера сообщений, вам нужно изменить только приведенную выше команду.
2. Закончив работу с проектом, вы можете закрыть Docker-контейнер — окружение вашей рабочей машины по-прежнему будет чистым.

Теперь импортируем Celery в наш Django-проект.

Создание Django-проекта

Рекомендую создать отдельное виртуальное окружение и работать в нем.

$ pip install django==3.1 $ django-admin startproject celery_django $ python manage.py startapp polls

Ниже представлена структура проекта.

├── celery_django │ ├── __init__.py │ ├── asgi.py │ ├── settings.py │ ├── urls.py │ └── wsgi.py ├── manage.py └── polls ├── __init__.py ├── admin.py ├── apps.py ├── migrations │ └── __init__.py ├── models.py ├── tests.py └── views.py

Файл celery.py

Давайте приступим к установке и настройке Celery.

pip install celery==4.4.7 redis==3.5.3 flower==0.9.7 

Создайте файл celery_django/celery.py рядом с celery_django/wsgi.py.

 
""" Файл настроек Celery https://docs.celeryproject.org/en/stable/django/first-steps-with-django.html """ from __future__ import absolute_import import os from celery import Celery # этот код скопирован с manage.py # он установит модуль настроек по умолчанию Django для приложения 'celery'. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_django.settings') # здесь вы меняете имя app = Celery("celery_django") # Для получения настроек Django, связываем префикс "CELERY" с настройкой celery app.config_from_object('django.conf:settings', namespace='CELERY') # загрузка tasks.py в приложение django app.autodiscover_tasks() @app.task def add(x, y): return x / y
Файл __init__.py
Давайте продолжим изменять проект, в celery_django/__init__.py добавьте.
 
from __future__ import absolute_import, unicode_literals # Это позволит убедиться, что приложение всегда импортируется, когда запускается Django from .celery import app as celery_app __all__ = ('celery_app',)
Дополнение settings.py
Поскольку Celery может читать конфигурацию из файла настроек Django, мы внесем в него следующие изменения.
 
CELERY_BROKER_URL = "redis://127.0.0.1:6379/0" CELERY_RESULT_BACKEND = "redis://127.0.0.1:6379/0"
Есть кое-что, о чем следует помнить.
При изучении документации Celery вы вероятно увидите, что broker_url — это ключ конфигурации, который вы должны установить для диспетчера сообщений, однако в приведенном выше celery.py:

app.config_from_object('django.conf: settings', namespace = 'CELERY') сообщает Celery, чтобы он считывал значение из пространства имен CELERY , поэтому, если вы установите просто broker_url в своем файле настроек Django, этот параметр не будет работать. Правило применяется для всех ключей конфигурации в документации Celery.
Некоторые конфигурационные ключи различаются между Celery 3 и Celery 4, так что, пожалуйста, загляните в документацию при настройке.

Отправка заданий Celery
После завершение работы с конфигурацией все готово к использованию Celery. Мы будем запускать некоторые команды в отдельном терминале, но я рекомендую вам взглянуть на Tmux, когда у вас будет время.
Сначала запустите Redis-клиент, потом celery worker в другом терминале, celery_django — это имя Celery-приложения, которое вы установили в celery_django/celery.py.
$ celery worker -A celery_django --loglevel=info -------------- celery@DESKTOP-111111 v4.4.7 (cliffs) --- ***** ----- -- ******* ---- Linux-4.4.0-19041-Microsoft-x86_64-with-glibc2.27 2021-03-15 15:03:44 - *** --- * --- - ** ---------- [config] - ** ---------- .> app: celery_django:0x7ff07f818ac0 - ** ---------- .> transport: redis://127.0.0.1:6379/0 - ** ---------- .> results: redis://127.0.0.1:6379/0 - *** --- * --- .> concurrency: 4 (prefork) -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker) --- ***** ----- -------------- [queues] .> celery exchange=celery(direct) key=celery [tasks] . celery_django.celery.add 
Далее запустим приложение в новом терминале, которое поможет нам отслеживать Celery-задачу (я расскажу об этом чуть позже).
$ flower -A celery_django --port=5555 [I 210315 16:11:39 command:135] Visit me at http://localhost:5555 [I 210315 16:11:39 command:142] Broker: redis://127.0.0.1:6379/0 [I 210315 16:11:39 command:143] Registered tasks: ['celery.accumulate', 'celery.backend_cleanup', 'celery.chain', 'celery.chord', 'celery.chord_unlock', 'celery.chunks', 'celery.group', 'celery.map', 'celery.starmap', 'celery_django.celery.add'] [I 210315 16:11:39 mixins:229] Connected to redis://127.0.0.1:6379/0 
Затем откройте http://localhost:5555/. Вы должны увидеть информационную панель, на которой отображаются детали выполнения рабочего процесса Celery.
Теперь войдем в Django shell и попробуем отправить Celery несколько задач.
$ python manage.py migrate $ python manage.py shell . >>> from celery_django.celery import add >>> task = add.delay(1, 2)
Рассмотрим некоторые моменты:

Мы используем xxx.delay для отправки сообщения брокеру. Рабочий процесс получает эту задачу и выполняет ее.
Когда вы нажимаете клавишу enter для ввода task = add.delay(1, 2) , кажется, что команда быстро завершает выполнение (отсутствие блокировки), но метод добавления все еще активен в рабочем процессе Celery.
Если вы проверите вывод терминала, где был запущен Celery, то увидите что-то вроде этого:

[2021-03-15 15:04:32,859: INFO/MainProcess] Received task: celery_django.celery.add[e1964774-fd3b-4add-96ff-116e3578de de] [2021-03-15 15:04:32,882: INFO/ForkPoolWorker-1] Task celery_django.celery.add[e1964774-fd3b-4add-96ff-116e3578dede] s ucceeded in 0.013418699999988348s: 0.5 
Рабочий процесс получил задачу в 15:04:32, и она была успешно выполнена.
Думаю, теперь у вас уже есть базовое представление об использовании Celery. Попробуем ввести еще один блок кода.
>>> print(task.state, task.result) SUCCESS 0.5 
Затем давайте попробуем вызвать ошибку в Celery worker и посмотрим, что произойдет.
>>> task = add.delay(1, 0) >>> type(task) celery.result.AsyncResult >>> task.state 'FAILURE' >>> task.result ZeroDivisionError('division by zero') 
Как видите, результатом вызова метода delay является экземпляр AsyncResult.
Мы можем использовать его следующим образом:

Проверить состояние задачи.
Узнать возвращенное значение (результат) или сведения об исключении.
Получить другие метаданные.

Мониторинг Celery с помощью Flower
Flower позволяет отобразить информацию о работу Celery более наглядно на веб-странице с дружественным интерфейсом. Это значительно упрощает понимание происходящего, поэтому я хочу обратить внимание на Flower, прежде чем углубиться в дальнейшее рассмотрение Celery.
URL-адрес панели управления: http://127.0.0.1:5555/. Откройте страницу задач — Tasks.

При изучении Celery довольно полезно использовать Flower для лучшего понимания деталей.
Когда вы развертываете свой проект на сервере, Flower не является обязательным компонентом. Я имею в виду, что вы можете напрямую использовать команды Celery, чтобы управлять приложением и проверять статус рабочего процесса.
Заключение
В этой статье я рассказал об основных аспектах Celery. Надеюсь, что после прочтения вы стали лучше понимать процесс работы с ним. Исходный код проекта доступен по ссылке в начале статьи.
Celery: начинаем правильно

10 Фев. 2016 , Python, 168594 просмотров, Celery Best Practices: practical approach

В этой статье мне хотелось бы поделиться с читателями своим опытом работы с таким замечательным инструментом в Python как Celery. Celery это ничто иное как распределённая очередь заданий, реализованная на языке Python. На момент написания этой статьи, самой последней версией является 3.1.20. Неосведомлённый читатель может не знать для чего вообще нужна система очередей задач наподобие Celery, поэтому кратко поясню этот момент.
Что такое Celery и зачем оно нам?
Часто ли вам приходилось сталкиваться с типовыми задачами в веб-приложениях вроде отправки электронного письма посетителю или обработки загруженных данных. Чаще всего такого рода манипуляции не требуют участия конечного пользователя вашего проекта, то есть их можно выполнять в фоновом режиме. Те из нас, кто реализует выполнение этих задач в одном из процессов веб-сервера, "тормозят" тем самым его работу, увеличивая время отклика и ухудшают user experience.
В данной заметке я опущу вводную информацию по установке и настройке Celery в вашем проекте. Кстати, Celery из коробки умеет работать с Django. Ранее был отдельный python пакет, соединяющий Django и Celery,именовался он django-celery. Сейчас он заброшен, так как последнее обновление было более года назад. Стоит отметить, что django-celery не работает Django 1.9 из-за изменений в работе cache backend. Исправленную версию можно посмотреть в моём форке. Одной из удобных фич django-celery является интеграция с Django Admin по части управления periodic tasks.
Советы по работе с Celery
Не используйте базу данных в качестве broker/backend
Брокер отвечает за передачу сообщений (задач) между так называемыми исполнителями (workers). Проблема использования базы данных заключается в её ограничениях - она просто не предназначена для этого. Дело в том, что с ростом количества исполнителей, нагрузка на базу будет только возрастать, а учитывая тот факт, что каждый worker имеет ещё ряд потоков, ситуация может стать катастрофической даже при малых нагрузках. Всё это приведёт к бутылочному горлышку в виде затыка на I/O, потере задач, а возможно и неоднократному их исполнению (два воркера могут получить одну и ту же задачу на исполнение). Отличным production-ready решением является использование RabbitMQ или Redis для этой роли.
Бэкэнд в случае с Celery выступает в качестве хранилища результатов выполнения задач (task). Одной из причин создания django-celery как раз являлась возможность подключения БД для сохранения результатов. Признаюсь, что в самом начале работы с Celery я неоднократно в проектах использовал этот подход. Пожалуйста, не повторяйте мою ошибку. С ростом нагрузки на приложение проблемы будут расти словно грибы после дождя (более того, "из коробки" celery не чистит базу от "устаревших" результатов) . Правда тут есть нюансы касательно вашего приложения. Об этом читайте ниже. Production-ready решением для роли backend неплохо зарекомендовал себя демон memcached. Пользуемся более 2-х лет, проблем ни разу не было.
Разделяйте задачи по очередям
Это очень важный момент. По мере развития вашего приложения, в проекте будут появляться критичные для выполнения задачи: проверка статуса платежа, формирование отчёта, отправка электронных писем и так далее. Терять их недопустимо. Если все задачи складировать в одну очередь, то в один прекрасный момент она может забиться, поставив под угрозу выполнение критически важного кода. Мой подход: разделяйте очереди по приоритетам.
Несомненно очередей может быть больше, тут всё на усмотрение разработчика и архитектуры его приложения.
В базовых настройках Celery это выглядит следующим образом:
CELERY_QUEUES = ( Queue('high', Exchange('high'), routing_key='high'), Queue('normal', Exchange('normal'), routing_key='normal'), Queue('low', Exchange('low'), routing_key='low'), ) CELERY_DEFAULT_QUEUE = 'normal' CELERY_DEFAULT_EXCHANGE = 'normal' CELERY_DEFAULT_ROUTING_KEY = 'normal' CELERY_ROUTES = < # -- HIGH PRIORITY QUEUE -- # 'myapp.tasks.check_payment_status': , # -- LOW PRIORITY QUEUE -- # 'myapp.tasks.close_session': , > 
В данном конкретном примере объявлена очередь по-умолчанию под названием normal. То есть задачи явно не указанные в списке будут автоматически распределены в эту очередь. В high попадает задача под названием check_payment_status, а в low задача close_session.
Запускать исполнителей Celery для этих очередей необходимо следующим образом:
celery worker -E -l INFO -n worker.high -Q high celery worker -E -l INFO -n worker.normal -Q normal celery worker -E -l INFO -n worker.low -Q low 
Здесь мы явно задаём имена исполнителей и названия очередей в которых необходимо мониторить задачи на исполнение.
ВАЖНО! Если вы явно указали для задачи очередь в которую ей нужно будет падать, и при этом запустили одного из исполнителей Celery без явного указания очереди, например вот так:
celery worker -E -l INFO -n worker.whatever 
То при наступлении ситуации, когда все исполнители очереди high будут заняты, Celery автоматически перенаправит новую задачу исполнителям без конкретной очереди. Поэтому при использовании раздельных очередей задач, не запускайте исполнителей без указания для них явного наименования очереди.
Логгируйте ошибки
Логгирование ошибок и своевременный их анализ это основа надёжных приложений. Очень важно иметь полную картину происходящего внутри вашего кода. По-умолчанию Celery все ошибки пишет в stderr, а прочая информация, связанная с исполнением попадает в stdout. Контролировать вывод ошибок можно через стандартный python logging, достаточно повесить свой handler на logger под названием "celery". Практика развёртывания боевых приложений, использующих Celery, показывает, что в качестве процесс-менеджера используют supervisord. В его настройках можно задавать путь до файла в который он будет складировать всю информацию, генерируемую демоном. Но вручную анализировать текстовые логи на предмет ошибок неудобно и неэффективно. Лично я использую для этих целей Sentry. Вот как выглядит у меня logging config:
CELERYD_HIJACK_ROOT_LOGGER = False LOGGING = < 'handlers': < 'celery_sentry_handler': < 'level': 'ERROR', 'class': 'core.log.handlers.CelerySentryHandler' >>, 'loggers': < 'celery': < 'handlers': ['celery_sentry_handler'], 'level': 'ERROR', 'propagate': False, >, > > 
Важной опцией здесь является наличие CELERYD_HIJACK_ROOT_LOGGER = False. По-умолчанию значение этой переменной является True, что позволяет celery "перекрывать" все ранее объявленные кастомные обработчики logging.
При указанном выше подходе нет необходимости дополнительно в коде задач (task) логгировать ошибки/исключения отдельно. О том что такое Sentry, для чего оно используется и как его настроить я напишу отдельную статью немного позже.
Пишите задачи маленькими
При написании задач старайтесь придерживаться принципа минимализма кода. То есть не нужно в самом celery task описывать бизнес логику задачи. Например, если вам необходимо генерировать и отправлять отчёт, то не нужно в самом task писать код генерации и отправки. Разбейте его на 3 части:

Код генерации отчёта
Код отправки письма
Задача (task) по выполнению этих действий

from .utils import generate_report, send_email @app.task(bind=True) def send_report(): filename = generate_report() send_email(subject, message, attachments=[filename]) 
Это, во-первых, позволит легче читать код (есть явное разделение на подзадачи). Во-вторых, тестировать такой код намного легче (привет модульным тестам!). В-третьих, отлавливать ошибки также будет намного легче и прозрачнее.
"Гасите" задачи вовремя
Явно указывайте лимит на выполнение задачи. Это можно сделать несколькими способами:

Через декоратор @app.task, передавая soft_time_limit, time_limit.
Глобально задать таймлимит при запуске исполнителя (worker), передав ему соответствующие аргументы (их можно найти в документации к Celery). В этом случае для всех задач, попадающих в заданную очередь будет один и тот же таймлимит.

Указание таймлимита очень важно, так как в некоторых случаях его отсутствие попросту приведёт к "зависанию" исполнителя при выполнении неоднозначных задач (требующих длительного времени, коннект к внешнему сервису и так далее).
Не храните результаты исполнения без необходимости
В большинстве случаев результат выполнения вашей задачи вам не нужен (например, если происходит отправка письма). В такой ситуации вам нет необходимости хранить что-то. Если ваши задачи полностью попадают в эту категорию, то в настройках Celery можно задать глобальный параметр CELERY_IGNORE_RESULT = True, который будет игнорировать результат исполнения всех ваших task-функций.
Используйте Flower для мониторинга исполнения задач
Всегда используйте Flower при работе с Celery. Всегда! Данный инструмент это небольшое веб приложение, написанное с использованием микрофреймворка Flask, а также Tornado для поддержки веб-сокетов. Flower позволяет вам всегда быть в курсе того как исполняются ваши задачи. Немного скриншотов:


Не поленитесь и потратьте время на его изучение. Оно окупится многократно!
Не передавайте ORM объекты в качестве аргументов
Я пару раз попадался на этом хитром трюке, который потрепал мне изрядно нервы. Рассмотрим вот такой код:
from .models import Profile @app.task(bind=True): def send_notification(profile): send_email(profile.user.email, subject, message_body) profile.notified = True profile.save() def notify_user(): profile = Profile.objects.get(id=1) check_smthng() send_notification.delay(profile) profile.activated = True profile.save() 
Не самый лучший пример для демонстрации побочного эффекта при передаче ORM объекта, но всё же. В данной ситуации код, описанный в send_notification, сохранит объект, изменив лишь notified = True, но activated останется по-прежнему равен False. Лучшим решением будет передача идентификатора объекта в базе данных, а в самой task функции необходимо непосредственно обращаться к объекту через его id.
BROKER_TRANSPORT_OPTIONS и visibility_timeout
При использовании Celery нередко приходиться прибегать к помощи отложенных задач, используя apply_async и передавая аргументы eta или countdown. Но делать это нужно осторожно, так как даже здесь нас поджидают "подводные камни". О чём речь? Очень часто у разработчиков, начинающих использовать очередь задач вроде Celery, происходят аномалии вроде выполнения одного и того же таска несколькими воркерами одновременно. Согласитесь, нежелательный сценарий. Так может происходить по причине того, что время, через которое должна выполниться задача, превышает visibility_timeout. По умолчанию для Redis этот параметр равен 1 часу. То есть если вы укажете выполнение задачи через 2 часа, то демон celery подождёт 1 час, поймёт, что никто из доступных воркеров не откликнулся и насильно назначит всем воркерам её выполнение при наступлении дедлайна (eta/countdown). Поэтому не забывайте про этот параметр, если вы собираетесь использовать механизмы eta/countdown/retry, задайте visibility_timeout равным самому длительному eta/countdown в вашем проекте. Подробнее можно почитать тут.
UPD: С недавних пор у блога появился свой Telegram канал, где я стараюсь делиться со своими подписчиками интересными находками из сети на тему разработки программного обеспечения и смежных с этой областью материалов.
Long-running tasks
Старайтесь не использовать Celery для выполнения долгих задач. На этот аргумент есть ряд причин:

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

Если нет возможности использовать что-то другое, то при работе с long-running tasks в Celery знайте следующее:
По-умолчанию 1 воркер процесс будет забирать из очереди 4 задачи за раз. Это особенно актуально знать, если Celery масштабируется на кластере через центрального брокера. То есть, если у вас 3 отдельные машины и на каждой крутится по 10 воркеров на очередь, то каждая машина будет забирать по 40 задач. Отсюда очевидно возникает проблема равномерного распределения задач по кластеру. Такое поведение оправдано в некоторых случаях, т.к. оно уменьшает количество обращений к брокеру, увеличивая производительность при выполнении небольших тасков. Чтобы изменить это, переопределите параметр CELERYD_PREFETCH_MULTIPLIER. Например:
CELERYD_PREFETCH_MULTIPLIER = 1
Долгоживущие процессы имеют тенденцию к пожиранию памяти, но вот назад её зачастую не возвращают, поэтому в контексте использования Celery с ними иногда имеет смысл перезагружать воркеры после выполнения заданного количества тасков. За это отвечает параметр CELERYD_MAX_TASKS_PER_CHILD
CELERYD_MAX_TASKS_PER_CHILD= 1
Настройка выше будет перезагружать воркер-процесс после выполнения 1 таска.
Полезные ссылки

Документация Celery
Celery: Distributed Task Queue
Flower: Celery task monitoring
Пакет django-celery
Python RQ: очередь задач на базе Redis

Celery: проясняем неочевидные моменты
Начнем с небольшой философии на тему разработки. Вопрос простой: должны ли мы сначала до последней страницы изучить документацию, а затем, когда мы, предположительно, полностью поймем библиотеку изнутри, начать использовать ее в своем коде? Или мы должны сначала использовать ее, поиграться с ней, прежде чем возвращаться и читать документацию.
Celery на самом деле полна подводных камней. Отчасти потому, что внутри происходит работа с параллельными процессами, потоками . и большую часть времени такие детали скрываются. Зачастую разработчику не нужно думать о них, и, следовательно, контакта со всем этим почти нет. И отчасти поэтому, для разработчика Celery иногда ведет себя самым неожиданным образом. Поэтому в нашем случае чтение документации все-таки необходимо.
Обработчики и Брокеры (Workers & Brokers)
Для начала, нужно объяснить некоторые основные понятия, которые используются в Celery.
Celery — это "Очередь задач" ( Task queue ). Да, для меня тоже было неизвестно, что это реальный термин. Мне казалось, что это описание того, что это такое в принципе — очередь задач, которые в конечном итоге будут выполнены. Итак, Celery — это программа, которая отслеживает задачи ( tasks ), которые необходимо выполнить, и в которой есть набор обработчиков ( workers ), которые будут выполнять эти задачи. Основной смысл в том, что она (программа) может выполнять несколько задач параллельно и что она не блокирует поставщиков ( producers ) этих самых задач.
Celery на самом деле не хранит все эти задачи в памяти. Для хранения задач есть отдельный сервис, называемый брокером сообщений ( message broker ), который по сути своей является очередью. Обычно это либо Redis, либо RabbitMQ. Т.е. Celery следит за тем, что происходит в очереди, но хранится она внутри Redis/RabbitMQ.
При запуске Celery создается 1 обработчик.
celery -A tasks worker
Этот обработчик является главным процессом ( supervisor process ), который будет порождать дочерние процессы или потоки, которые в свою очередь будут выполнять задачи. По умолчанию главный обработчик будет создавать дочерние процессы, а не потоки, и он создаст столько одновременных дочерних процессов, сколько ядер у процессора. Главный процесс будет следить за тем, что происходит с задачами и процессами/потоками, но он не будет запускать сами задачи. Эта группа дочерних процессов или потоков, которая ожидает выполнения задач, называется пулом выполнения ( execution pool ) или пулом потоков ( thread pool ).
Очереди (Queues)
Да, тут намеренно используется множественное число для очередей, потому что существует несколько видов очередей ����‍⚗️.
Прежде всего, существует главная очередь ( main queue ), которая принимает задачи от поставщиков ( producers ) по мере их поступления и передает обработчикам по мере их запроса. По умолчанию есть только одна такая очередь. Все обработчики принимают задачи из одной очереди. Но вы также можете указать несколько таких очередей и назначить конкретные обработчики на определенные очереди. Очередь по умолчанию называется celery .
Чтобы просмотреть первые 100 задач в очереди в Redis, выполните:
redis-cli lrange celery 0 100
Эти очереди сильно напоминают FIFO (First In First Out), но это не совсем так. Задачи, которые сначала помещаются в очередь, первыми удаляются из очереди, НО они не обязательно выполняются первыми.
Когда обработчики извлекают новые задачи из очереди, они обычно (по умолчанию) берут не столько задач, сколько у них есть процессов, они берут больше. По умолчанию формула для количества взятых задач такая: 
# n_processes — количествово потоков/процессов n_tasks = 4 * n_processes
Делается это для экономии времени. Взаимодействие с брокером занимает какое-то время, и если задачи, которые необходимо выполнить, выполняются быстро, то обработчики будут запрашивать дополнительные задачи снова, и снова, и снова. Чтобы избежать этого, они запрашивают в X раз больше задач, чем у них есть процессов, за это отвечает параметр worker_prefetch_multiplier (прим: этот момент хорошо описан в документации).
Но ведь есть задачи, которые никогда не попадают в главную очередь и все равно выполняются. Как это возможно, спросите вы меня? Задавая себе и Google один и тот же вопрос, хочу сообщить вам, что Google очень мало что смог сказать по этому поводу. Нашлись только обрывки информации. Но, посидев с Celery и Redis несколько часов (или дней??), обнаружилось следующее.
Задачи с ETA никогда не помещаются в главную очередь. Они помещаются во что-то среднее между очередью и списком “неподтвержденных задач”, и называется эта очередь unacked (прим: сокращение от слова "unacknowledged"). Согласитесь, что очень легко пропустить что-то с названием unacked , когда вы пытаетесь понять, как и куда некоторые задачи только что исчезли. Итак, примечание для следующего раза, когда мне или вам нужно будет что-то назвать: все названия, с которыми встречается пользователь, должны быть прописаны полностью.
Итак, что представляют собой ETA задачи? Это запланированные задачи. ETA расшифровывается как "estimated time of arrival" (“предполагаемое время прибытия”). Все задачи, для которых указано ETA или обратный отсчет ( countdown ), например:
my_task.apply_async((1, 2), countdown=3) my_task.apply_async((1, 2), eta=tomorrow_datetime)
хранятся в unacked очереди. Сюда же попадают и задачи с ретраями, потому что при повторном выполнении задачи она повторяется через определенное количество секунд, а это означает, что у нее есть ETA.
Посмотреть, какие задачи находятся в очереди ETA в Redis, можно вот так:
redis-cli HGETAL unacked
В ответ получите список из чередующихся ключей и значений:
1) "46165d9f-cf45-4a75-ace1-44443337e000" 2) "[ 

Задачи (Tasks)
Задачи иногда также называют сообщениями. По сути брокер сообщений - это нечто, что передает сообщения из одной системы в другую. В нашем случае сообщение представляет собой описание задачи: название (уникальный идентификатор), входные параметры, время ожидания, количество повторных попыток и тд.
В celery задача является классом. Таким образом, каждый раз, когда вы используете декоратор для функции (например, @shared_task ), чтобы сделать ее celery задачей, под капотом создается класс. Это означает, что у каждой задачи есть self , к которому добавляется множество атрибутов, например: name , request , status , priority , retries и многое другое. Если мы хотим получить доступ к этим атрибутам, то нужно указать параметр bind=True .
@shared_task(bind=True. ) def _send_one_email(self, email_type, user_id): . num_of_retries = self.request.retries . 
Подтверждение задач (Task Acknowledgment)
Ранее мы говорили, что, когда у обработчиков нет задач, они идут и получают еще несколько задач от брокера. Но не все так просто. Когда обработчик “берет” задачу, задача перемещается из главной очереди в unacked очередь. Задача полностью удаляется из брокера только после того, как обработчик подтвердит это. Это означает, что когда обработчик забирает себе очередную пачку задач, на самом деле в этот момент задачи только резервируются. Они помещаются в unacked очередь и другие обработчики их не возьмут. Если процесс обработчика умирает, то эти задачи становятся доступными для других обработчиков.
Итак, когда обработчик всё же подтверждает выполнение задачи? По умолчанию Celery предполагает, что опасно запускать задачу более одного раза, следовательно, подтверждение задачи происходит непосредственно перед ее выполнением. Вы можете изменить это, установив значение acks_late=True. В этом случае задача имеет небольшую вероятность быть запущенной более одного раза, если обработчик, выполняющий ее, умирает в середине выполнения. И под “умирает” буквально подразумевается умереть. Python Exception в коде задачи не приведет к смерти обработчика. Такая задача по-прежнему будет подтверждена, но ее состояние будет FAILURE . Должно произойти что-то из ряда вон выходящее, чтобы обработчик никогда не достиг момента self.confirmate() . И на самом деле это редкость. По этой причине, можно сказать, что значение параметра acks_late имеет мало значения.
ETA
Как уже упоминалось, ETA задачи . тяжело отыскать. Они никогда не попадают в главную очередь. Они сразу назначаются обработчику и помещаются в unacked очередь. Я подозреваю, что это не было сделано намеренно, чтобы задачи ETA немедленно назначались конкретному обработчику. Скорее это было просто следствием существующего кода. Задача ETA не может попасть в общую очередь, которая работает почти как FIFO. Единственное другое место для нее находится среди неподтвержденных задач, и в этом случае она должно быть зарезервирована одним обработчиком.
Интересно, что время ETA не является точным временем выполнения этой задачи. Вместо этого это самый ранний момент выполнения этой задачи. Как только наступит время ETA, задача должна дождаться освобождения обработчика.
Повторное исполнение задач (Retry Tasks)
Celery по умолчанию сама не делает ретраи для задач. Главным образом потому, что предполагается, что задачи не являются идемпотентными (прим: одинаковыми независимо от момента выполнения), и поэтому небезопасно запускать их более одного раза. Однако, в Celery есть функционал для ретраев задач, но он должен быть явно и отдельно настроен для каждой задачи.
Одним из способов запуска ретрая является вызов self.retry() в задаче. Что происходит после того, когда вы вызываете эту команду? Вычисляется время ETA, собираются некоторые новые метаданные, а затем задача отправляется брокеру, где она попадает в unacked очередь и назначается тому же обработчику, который уже выполнял эту задачу. Именно так ретрай-задачи становятся задачами ETA и, следовательно, никогда не отображаются в главной очереди брокера. Это очень изящная, но неожиданная схема. И опять же, Google очень мало что может сказать по этому поводу.
Подробнее о ретраях читайте в гайде по ретраям в Celery.
CPU, I/O ограничения и Процессы vs Потоки
Как мы уже говорили, по умолчанию Celery выполняет задачи в отдельных процессах, а не в потоках. Но вы можете заставить использовать потоки, стартуя обработчики либо с помощью --poll eventlet , либо с помощью --pool gevent . И eventlet, и gevent на самом деле создают гринлеты, а не потоки. Гринлеты (или зеленые потоки) похожи на потоки, но не являются ими, потому что по определению потоки управляются операционной системой. Гринлеты не полагаются на ОС для обеспечения поддержки потоков, вместо этого они эмулируют многопоточность. Они управляются в пространстве программы, а не в пространстве операционной системы. В любой данный момент нет переключения между потоками. Вместо этого гринлеты добровольно или явно передают управление друг другу в определенных точках вашего кода.
Если ваши задачи сильно загружают процессор, если они выполняют много вычислений (CPU-bound), то вам следует продолжать использовать процессы. Если, с другой стороны, ваши задачи в основном выполняют HTTP-запросы (I/O bound), то лучше использовать потоки. Причина этого заключается в том, что, пока ваша задача ожидает, пока HTTP-запрос вернет результат, она ничего не делает, она не использует процессор и, следовательно, не будет "возражать", если другой поток будет использовать его.
В Celery гораздо больше неочевидных моментов
Документация в Celery далека от идеала. Описание многих функций разбито на части и разбросано по всем страницам. Трудно найти детали реализации. Я не знаю, как Celery будет вести себя за пределами сценариев, которые создает разработчик. Конечно, после нескольких лет интенсивной работы можно было бы хорошо понять, как это работает, но Celery живет на периферии моей повседневной жизни. Celery ведет себя совершенно по-разному, когда находится на сервере и когда находится на моем компьютере. Я вижу, какие задачи были выполнены, но я не вижу, насколько хорошо они были выполнены. Это сложно измерить у того, что выполняется параллельно, в потоках и почти независимо от вашей программы. Я не доверяю Celery, я не верю, что правильно понимаю настройки, или, можно сказать, я не верю, что знаю, как их правильно настроить. Celery подобен приведению, которое приходит и уходит, иногда ломается, но в большинстве случаев просто работает. Надеюсь, Celery справится с теми задачами, которые мы ему поручаем, но если это не так, он будет работать так же тихо и мы ничего об этом не узнаем.
Конечно, можно сказать, что всё это можно было вычитать из официальной документации. Как бы да, но нет. И это доказывает нам автор статьи Ines Panker, которая провела много часов (дней?) на чтение документации, поиск в гугле, тыканье в python/celery/redis. Скажем же ей спасибо за это. А я надеюсь, что эта статья была полезна для вас и что у вас появилось больше понимания того, что вы используете.

Никита Шамаев
Подписывайтесь на мой телеграм-канал. Там еще больше постов про разработку/технологии/датасеты, и там вы не пропустите анонс следующей статьи ��
Использование Celery с Django для обработки фоновых задач
Когда это происходит, нужно провести различие между тем, что должно выполняться мгновенно (обычно в жизненном цикле HTTP-запроса), и что может произойти в конечном итоге, после запроса. Почему это необходимо? Ну, потому что, когда ваше приложение становится перегруженным трафиком, решения, подобные этому, начинают иметь значение.
Операции в веб-приложении могут быть классифицированы как критические или request-time операции и фоновые задачи, которые происходят за пределами цикла запрос-ответ. Они сопоставляются с описанными выше:

Должно произойти мгновенно: request-time операция
Должно произойти после ответа: фоновые задачи

Request-time операции могут выполняться в течение одного цикла запрос/ответ, не беспокоясь о том, что операция может получить тайм-аут или что у пользователя может быть плохое соединение. Общие примеры включают CRUD (создание, чтение, обновление, удаление) операций с базой данных и управление пользователями (процедуры входа/выхода).
Фоновые задачи отличаются, поскольку они обычно довольно трудоемки и склонны к сбою, главным образом из-за внешних зависимостей. Некоторые распространенные сценарии среди сложных веб-приложений включают:

Отправка уведомлений о подтверждении или рассылка сообщений
Ежедневное сканирование и скрапинг некоторой информации из разных источников и сохранение их
Анализ данных
Удаление ненужных ресурсов
экспорт документов/фотографии в различных форматах

Фоновые задачи являются основным направлением данного руководства. Наиболее распространенным шаблоном программирования, используемым для этого сценария, является Producer-Consumer архитектура.
Проще говоря, эта архитектура может быть описана следующим образом:

Producers создают данные или задачи.
Задачи помещаются в очередь, которая называется очередью задач.
Consumers несут ответственность за потребление данных или выполнение задач.

Обычно сonsumers загружают задачи из очереди в режиме first-out (FIFO) или в соответствии с их приоритетами. Потребителей также называют воркерами, и это термин, который мы будем использовать повсюду, поскольку он согласуется с терминологией, используемой обсуждаемыми технологиями.
Какие задачи можно обрабатывать в фоновом режиме? Задачи, которые:

Не являются существенными для основного функционала веб-приложения
Не могут выполняться в цикле запросо/ответ, поскольку они медленны (интенсивность ввода-вывода и.т.д.)
Зависят от внешних ресурсов, которые могут быть недоступны или не будут вести себя так, как ожидалось
Возможно, потребуют повторить попытку хотя бы один раз
Должны выполняться по графику

Celery является де-факто выбором для обработки фоновых задач в экосистеме Python/Django. Он имеет простой и понятный API, и он прекрасно сочетается с Django. Он поддерживает различные технологии для очереди задач и различные парадигмы для воркеров.
В этом уроке мы собираемся создать игровое веб-приложение Django (работающее с реальными сценариями), которое использует обработку фоновых задач.
Настройка
Предполагая, что вы уже знакомы с менеджером пакетов Python и виртуальными окружениями, давайте установим Django:
$ pip install Django
Я решил создать еще одно приложение для ведения блога. Приложение будет простым. Пользователь может просто создать учетную запись и без особых проблем может создать пост и опубликовать его на платформе.
Настройка Django проекта quick_publisher :
$ django-admin startproject quick_publisher
Давайте начнем разработку приложения:
$ cd quick_publisher
$ ./manage.py startapp main
При запуске нового Django проекта мне нравится создавать main приложение, которое содержит, помимо прочего, кастомную модель пользователя. Чаще всего Я сталкиваюсь с ограничениями модели User по умолчанию. Наличие пользовательской модели User дает нам гибкость.
# main/models.py 
from django.db import models 
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin, BaseUserManager 
class UserAccountManager(BaseUserManager): 
use_in_migrations = True 
def _create_user(self, email, password, **extra_fields): 
if not email: 
raise ValueError('Email address must be provided') 
if not password: 
raise ValueError('Password must be provided') 
email = self.normalize_email(email) 
user = self.model(email=email, **extra_fields) 
user.set_password(password) 
user.save(using=self._db) 
return user 
def create_user(self, email=None, password=None, **extra_fields): 
return self._create_user(email, password, **extra_fields) 
def create_superuser(self, email, password, **extra_fields): 
extra_fields['is_staff'] = True 
extra_fields['is_superuser'] = True 
return self._create_user(email, password, **extra_fields) 
class User(AbstractBaseUser, PermissionsMixin): 
REQUIRED_FIELDS = [] 
USERNAME_FIELD = 'email' 
objects = UserAccountManager() 
email = models.EmailField('email', unique=True, blank=False, null=False) 
full_name = models.CharField('full name', blank=True, null=True, max_length=400) 
is_staff = models.BooleanField('staff status', default=False) 
is_active = models.BooleanField('active', default=True) 
def get_short_name(self): 
return self.email 


Похожие публикации:

Как открыть revolut в россии 2022
Как работает else в циклах питон
Какие вопросы можно задать чат боту
Чего фаеаы