Celery что это

Celery: начинаем правильно

10 Фев. 2016 , Python, 168590 просмотров, 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 части:

1. Код генерации отчёта
2. Код отправки письма
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 для выполнения долгих задач. На этот аргумент есть ряд причин:

1. Процессы, живущие долго, потребляют память, но не освобождают её. Даже с учётом работы сборщика мусора. Такой механизм необходим, чтобы избежать фрагментации оперативной памяти.
2. 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» на русский

And of course, calcium is found in plant foods, especially salad vegetables (celery, cabbage, leaf lettuce) and herbs.

И конечно кальций содержится в растительных продуктах, особенно салатных овощах (сельдерей, капуста, листовой салат) и зелени.

Furthermore, celery seeds help regulate metabolism to prevent sleep disorders like insomnia.

Кроме того, семена сельдерея помогают регулировать обмен веществ, чтобы предотвратить нарушения сна, такие как бессонница.

Follow package directions if using a celery seed extract.
Следуйте указаниям по упаковке, если вы используете экстракт семян сельдерея.
Sesame or celery seeds might cook long enough when toasted since they are small.

Кунжут или семена сельдерея могут готовиться достаточно долго, когда их жарят, так как они маленькие.

Every morning Holmes drinks celery fresh to keep health.
Каждое утро Элизабет пьет фреши из сельдерея, чтобы быть здоровой.
Calcium contains cabbage, celery, dairy products.
Кальций содержат капуста, сельдерей, молочные продукты.

Drinking celery juice every morning is a new health trend that’s marketed as improving overall health and boosting weight loss.

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

It has a sweet taste similar to celery.
Они имеют сладковатый вкус, похожий на сельдерей.
It tells us that the fourth part of the calories that are in celery left.
А это означает, что четвёртая часть калорий, которые содержит сельдерей, остались.

This super food also contains important concentrations of plant hormones and the very special essential oils that give celery its characteristic smell.

Этот супер-продукт содержит важные концентрации растительных гормонов и специальных эфирных масел, которые придают сельдерею характерный запах.

In addition to this, wipe the skin with raw potato and drink celery juice, which can strengthen the immune system.

Вдобавок к этому протирайте кожу сырым картофелем и пейте сок сельдерея, который способен укрепить иммунитет.

Along with high water content, celery is rich in fiber.
Наряду с высоким содержанием воды, сельдерей богат клетчаткой.

Thanks to the powerful anti-inflammatory, antioxidant and antispasmodic properties, celery will help in relieving the symptoms quickly pleurisy.

Благодаря мощным противовоспалительным, антиоксидантным и спазмолитическим свойствами, сельдерей поможет быстро снимает симптомы плеврита.

Studies have proven that celery supplements the skeleton’s needs.
Было доказано, что сельдерей пополняет потребности нашего скелета.
In the first week after childbirth, she ate only soup of celery.
В первую неделю после родов она питалась только супом из сельдерея.
Do not forget about celery, he is also rich in these valuable substances.
Не нужно забывать и про сельдерей, он также богат этими ценными веществами.
Another reason why we must have celery, is that it is very useful to our eyes.

Еще одна причина, по которой мы должны есть сельдерей, заключается в том, что он очень полезен для наших глаз.

You can easily determine, because it is exactly what the substance that gives celery this special smell.

Вы легко определите, потому что это именно то вещество, которое придает сельдерею особый запах.
And in all publications the most powerful and effective means was called celery.
И во всех изданиях наиболее сильным и действенным средством назывался сельдерей.
Возможно неприемлемое содержание

Примеры предназначены только для помощи в переводе искомых слов и выражений в различных контекстах. Мы не выбираем и не утверждаем примеры, и они могут содержать неприемлемые слова или идеи. Пожалуйста, сообщайте нам о примерах, которые, на Ваш взгляд, необходимо исправить или удалить. Грубые или разговорные переводы обычно отмечены красным или оранжевым цветом.

Зарегистрируйтесь, чтобы увидеть больше примеров. Это просто и бесплатно
Ничего не найдено для этого значения.
Предложить пример
Больше примеров Предложить пример

Предложения, которые содержат celery

Новое: Reverso для Windows

Переводите текст из любого приложения одним щелчком мыши .

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

Результатов: 1589 . Точных совпадений: 1589 . Затраченное время: 81 мс

Помогаем миллионам людей и компаний общаться более эффективно на всех языках.

Оптимизация очередей через Celery и Flask: делаем это правильно

Если вы наткнулись на эту статью, скорее всего, вы знакомы с Flask и работаете над добавлением функции в свое веб-приложение, на выполнение которой уходит довольно много минут (если не больше). Возможно, вы хотите узнать, есть ли лучший или более быстрый способ сделать это. Да, именно так — такой способ есть, мы его подробно обсудим в этом справочном материале для всех любителей Python.

Професійний курс від mate.academy: Java.
Погрузьтеся у світ програмування.

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

Celery — это мощная очередь задач, которую можно использовать как для простых фоновых задач, так и для сложных многоэтапных программ и расписаний. Это руководство покажет вам, как использовать Celery с помощью Flask, но предполагает, что вы уже прочитали руководство «Первые шаги с Celery» в документации по Celery.

Celery — это отдельный пакет Python. Установите его из PyPI с помощью pip :

Потужний курс від laba: Управління ІТ-командами.
Максимізуйте продуктивність вашої ІТ-команди.

$ pip install celery

Введение

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

• Вызов стороннего API для получения некоторых данных на основе пользовательского ввода.
• Отправка электронного письма пользователю при регистрации.

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

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

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

Необходимые стартовые условия

Інтенсивний курс від skvot: Ретуш фотографій.
Перетворіть свої знімки у шедеври.

• Python 3.6+
• Virtualenv v20+
• Кроме того, ожидается промежуточное знание Python и Flask. Все остальное будет объяснено по ходу статьи.

Настройка проекта

Скачайте стартовый проект и настройте его с помощью следующих команд:

git clone -b step_1 https://github.com/raunaqness/flask_celery_tutorial.git cd flask_celery_tutorial # make virtualenv virtualenv v source v/bin/activate # install dependencies pip install -r requirements.txt # start server export FLASK_APP=app; python -m flask run

Откройте 127.0.0.1:5000 в браузере, и, если все работает нормально, вы должны увидеть надпись «Hello, world!».

Далее добавим маршрут, который будет содержать кнопку Button , при нажатии на которую будет запускаться имитация длительной задачи, например, отправка электронного письма, создание отчета в формате PDF, вызов API стороннего разработчика и т.д.

Мы сымитируем этот API с помощью time.sleep() , который заблокирует работу приложения на 15 секунд.

Откройте файл app.py и добавьте следующий блок кода.

# route that will show will simply render an HTML template @app.route("/tasks") def tasks(): return render_template("tasks.html") # route that will execute a long-running task @app.route("/long_running_task") def long_running_task(): # time in seconds time_to_wait = 15 print(f"This task will take seconds to complete. ") time.sleep(time_to_wait) return f"
The task completed in seconds!"

Убедитесь, что импортировали модуль времени, добавив следующее, вместе с утверждениями импорта в верхней части файла:

Цифровий курс від robotdreams: DevOps Engineer.
підходи для створення сучасних і масштабованих застосунків.

Далее создайте каталог с именем templates в корне проекта. Внутри него создайте новый файл tasks.html и добавьте в него следующее:

      [email protected]/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-1BmE4kWBq78iYhFldvKuhfTAU6auU8tT94WrHftjDbrCEXSU1oBoqyl2QvZ6jIW3" crossorigin="anonymous" />  
Trigger Long Running Task 
   

Структура (дерево) вашего проекта должна выглядеть примерно так:

code ├── __pycache__ │ └── app.cpython-38.pyc ├── app.py ├── requirements.txt └── templates └── tasks.html 2 directories, 4 files

Вернитесь в терминал, остановите и снова запустите сервер Flask, затем откройте 127.0.0.1:5000/tasks в браузере. Вы должны увидеть страницу tasks.html с единственной кнопкой.

Теперь, когда вы нажмете на кнопку Trigger Long-Running Task, произойдет переход к маршруту /long_running_task , который выполнит функцию def long_running_task() , определенную в файле app.py .

Обратите внимание, что страница будет находиться в состоянии «загрузки» в течение 15 секунд, так что ваше приложение застрянет в этом состоянии и не сможет выполнить никакую другую операцию, пока не завершится текущая.

Через 15 секунд вы должны увидеть выполненную задачу и ожидаемый ответ в браузере.

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

Теперь давайте посмотрим, как мы можем использовать Celery для выполнения этой задачи в фоновом режиме. Если у вас возникли проблемы, вы можете посмотреть текущее состояние вашего проекта вот здесь.

Настройка Celery и Redis

Вы уже установили пакет Celery python в начальной настройке. Чтобы подтвердить установку пакета, вы можете запустить pip freeze в окне терминала с активированным virtualenv , чтобы увидеть все установленные пакеты.

Далее необходимо установить Redis Server на локальной машине. Официальную инструкцию по установке можно найти вот здесь.

Теперь давайте настроим Celery.

Начало работы с Celery

Создайте новый файл в корне проекта под названием celery_utils.py . Он будет использоваться для инициализации экземпляра приложения Celery, аналогично тому, как мы инициализируем приложение Flask в app.py . Добавьте в файл следующий код:

from celery import Celery # celery config CELERY_BROKER_URL = 'redis://localhost:6379' CELERY_RESULT_BACKEND = 'redis://localhost:6379' # initialize celery app def get_celery_app_instance(app): celery = Celery( app.import_name, backend=CELERY_BROKER_URL, broker=CELERY_BROKER_URL ) celery.conf.update(app.config) class ContextTask(celery.Task): def __call__(self, *args, **kwargs): with app.app_context(): return self.run(*args, **kwargs) celery.Task = ContextTask return celery

Вот краткое объяснение:

• Импортируется пакет Celery python.
• Определяется функция get_celery_app_instance , которая возвращает экземпляр Celery, который, в свою очередь, требует следующих параметров для инициализации:
  • name: это имя рабочего Celery.
  • backend : это URL используемого бэкенда, которым в данном случае является Redis, а URL хоста определяется переменной CELERY_BROKER_URL .
  • брокер: аналогично бэкенду, необходимо определить URL брокера, который также является сервером Redis.

  Далее, давайте используем Celery для определения долгоиграющей задачи. Внесите следующие изменения в app.py :

  Добавьте следующее рядом с операторами импорта.

  # importing function to get celery app instance from celery_utils import get_celery_app_instance

  Добавьте следующее после утверждений, инициализирующих приложение Flask:

  # celery app instance celery = get_celery_app_instance(app)

  Затем добавьте следующий блок кода в нижнюю часть файла:

  # celery tasks @celery.task def sending_email_with_celery(): print("Executing Long running task : Sending email with celery. ") time.sleep(15) print("Task complete!")

  Здесь мы просто определили функцию с именем sending_email_with_celery() , которая будет имитировать отправку электронного письма, которая может занять 15 секунд.

  Однако для того, чтобы эта функция выполнялась как фоновая задача, в строке чуть выше определения функции добавлен декоратор @celery.task .

  Наконец, определите маршрут для запуска этой функции:

  # route to trigger celery task @app.route("/long_running_task_celery") def long_running_task_celery(): # function.delay() is used to trigger function as celery task sending_email_with_celery.delay() return f"Long running task triggered with Celery! Check terminal to see the logs. "

  В этом блоке кода мы определяем маршрут /long_running_task_celery , который запускает функцию как задачу Celery.

  Обратите внимание, что функция вызывается с помощью метода delay(). Это указывает на то, что мы хотим запустить эту функцию как задачу Celery, а не как обычную функцию Python.

  Наконец, чтобы увидеть это в действии, давайте добавим еще одну кнопку в файл tasks.html для запуска этой функции.

   
  Trigger Long Running Task 
    
  Trigger Long Running Task with Celery 

  Пришло время увидеть его в действии!

  Убедитесь, что у вас запущен сервер Flask в окне терминала.

  

  В другом окне терминала перейдите ( cd ) в корень проекта и выполните следующую команду для запуска Celery worker .

  celery -A app.celery worker --loglevel=info

  

  Откройте 127.0.0.1:5000/tasks в браузере, где вы должны увидеть две кнопки:

  • Запуск долго выполняющейся функции с помощью Python.
  • Запуск долго выполняющейся функции с помощью Celery.

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

  Теперь, если вы нажмете на кнопку Trigger Long-Running Task with Celery, вы увидите, что страница мгновенно перенаправляется на маршрут /long_running_task_celery , а в окне браузера появится ожидаемый результат.

  

  В фоновом режиме выполнение функции осуществляется Celery. Чтобы посмотреть журнал выполнения функции, переключитесь в окно терминала, в котором вы запустили Celery worker . Оно должно выглядеть примерно так:

  

  Заключение

  Вот и все! Теперь вы знаете, как настраивать и запускать длительно выполняющиеся задачи с помощью Celery в вашем веб-приложении Flask.

  Вот краткий обзор того, что мы обсудили выше. Чтобы запустить функцию как задачу Celery, вам необходимо:

  • Импортируйте экземпляр приложения Celery в ваш файл.
  • Добавьте декоратор @celery.task поверх определения функции.
  • Запустите функцию, используя метод function_name.delay() .

  Celery & Flower: построение и настройка очередей

  Сегодня поговорим о Celery — популярном инструменте для создания распределенных очередей задач. Такие очереди позволяют разгрузить работу в другой процесс и выполнять часть задач в фоновом режиме (если они не зависят от дальнейших действий пользователей), пока приложение выполняет другие задачи. Это ускоряет время отклика сайта или программы и улучшает user experience. Также поговорим о Flower — с его помощью мы отслеживаем воркеры Celery и прогресс по таскам.

  Что такое Celery

  Коротко о том, что собой являет Celery:

  • бесплатная open source библиотека на Python;
  • распределённая очередь задач: задачи могут распределяться по нескольким машинам, воркерам, ядрам процессора;
  • асинхронное выполнение задач: некоторые таски, попадающие на воркер, могут выполняться отдельно от основного потока;
  • задачи по расписанию: можно прописать расписание регулярных задач, и они будут выполняться ежеминутно, ежечасно или ежедневно.

  Для чего используется Celery

  Celery зарекомендовал себя как прекрасный инструмент для:

  • отложенного выполнения задач. Не все задачи требуют мгновенного выполнения: например, когда мы генерируем какое-то количество рассылок, мы не должны ожидать на фронте, пока они отправятся. Вместо этого получим результат типа success и сообщение, что задачи поставлены в очередь и начнут исполняться в определенный момент;
  • распределения вычислений между серверами. Можем отдать часть задач на сторонний сервер, чтобы не выполнять все таски на одной машине и не нагружать основной сервер. Для нас он более важен и должен справляться с нагрузкой трафика.

  

  Как работает Celery

  Рассмотрим коротко, из чего состоит архитектура Celery.

  1. В Celery задачи создают так называемые “продюсеры” (producers). Ими может выступать Celery beat (cron), который будет генерировать таски ежеминутно, ежечасно — как мы запрограммируем, либо же задачи будут генерироваться из приложения (не обязательно Django) по событию: когда пользователь что-то нажал на фронте, на бэкенде сгенерировалась задача.
  2. Таски попадают в очередь в брокер (Redis или RabbitMQ). Мы используем Redis, так как его легче настроить, и он лучше справляется с простыми задачами. В брокере можно организовать несколько очередей и присвоить каждой приоритет, исходя из важности.
  3. Consumers — собственно, воркеры (workers) — это исполнители задач, которые забирают и выполняют таски из очередей в брокере. Для каждого воркера очередь настроена индивидуально. Воркеры могут возвращать или не возвращать результаты. Если результаты выполнения задачи нам не нужны, можем их игнорировать. Если нужны, они будут храниться в бэкенде Celery в виде данных типа success и hash и информации о времени (длительности) выполнения задачи. Как бэкенд также может использоваться Redis.

  Лучшие практики в использовании Celery

  Делимся нашими наблюдениями и выводами, сделанными в процессе работы с очередями задач в Celery.

  Не использовать в качестве брокера/бэкенда БД

  База данных не совсем для этого предназначена и имеет свои ограничения: она не настолько быстрая и располагается, в отличие от Redis, не в оперативной памяти, а на жестком диске. Даже если это SSD, все равно требуется гораздо больше времени, чтобы ее использовать.

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

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

  Не передавать готовый ORM-объект или большие JSON/сериализованные объекты в задачу

  Во-первых, все это хранится в брокере в то время, пока задача находится в очереди (в Redis, который “висит” в оперативке) и забивает память. Во-вторых, если мы получили ORM-объект, к тому моменту, когда задача дойдет до выполнения, данные в этом объекте уже могут быть неактуальными. То есть, нужно сделать повторный запрос к БД, получить обновленные данные и их обрабатывать.

  Устанавливать лимиты на выполнение задач

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

  Использовать мелкие таски с минимальным функционалом

  Это полезно для модульности, сокращает количество кода (многомодульность), и задачи не висят подолгу на исполнении.

  Таскам лучше возвращать информативный результат

  Если мы обрабатываем и храним результаты выполнения задач, по ним должно быть понятно, что происходит с тасками. Конечно, это не должен быть большой объект или JSON, но и не простой ответ true or false.

  Логирование

  В момент запуска Redis, где хранятся результаты выполнения задач, эти данные могут быть утеряны. Если мы ведем логи, все сводится в определенный файл, и при необходимости можно проследить, что происходило с тасками не только по результатам, но и в процессе выполнения.

  Мониторить таски через веб-интерфейс Flower

  Это must have. Для доступа к мониторингу достаточно браузера — не нужно заходить в консоль, на сервер, при этом управление доступом в Flower гораздо проще и визуально понятнее.

  Что такое Flower и зачем его использовать

  При работе с Celery рекомендуем использовать Flower — небольшое веб-приложение, главными возможностями которого являются:

  • мониторинг в real-time (графики, поиск, статистика). Flower написан на Tornado и работает через веб-сокеты, данные в нем обновляются в реальном времени. Можем просматривать графики, искать по таскам, смотреть статистику. Однако Flower не сохраняет ранее выполненные таски (экономия памяти), и чтобы строился какой-то график, нужно постоянно держать его в браузере.

  

  • Управление воркерами : перезапуск, установка лимитов, количество задач, которые он обрабатывает за один раз.
  • Управление очередями : можем перебрасывать очереди между воркерами, задавать приоритетность. Если не ожидаем никаких высокоприоритетных задач, можем отключить эту очередь из воркера. Пример создания очередей в Flower:

  

  В настройках мы начали использовать autoscale: если не прописывать автоскейлинг или concurrency, то количество субпроцессов будет равно числу ядер. И это не очень хорошо, потому что в таком случае выполнение задач и обработка запросов веб-сервера будут “съедать” все ядра, и сайт или приложение будет подтормаживать.

  

  Процесс обработки задач можем увидеть на вкладке Tasks. Тут есть сортировки по состояниям, времени выполнения, можно посмотреть сами таски, есть трейсбеки (traceback), где отображается код и информация об ошибках, если задача упадет.

  

  Flower у нас подключен к брокеру и к бэкенду (Redis), но данные по таскам он хранит в себе. Если мы перезагрузим Redis, то результаты выполнения текущих задач все равно будут храниться в процессе Flower до момента его перезапуска. В то же время, если мы перезапустим Flower, то он обновится и не будет искать в Redis, какие таски были выполнены и какие еще ожидают выполнения.

  Все параметры, относящиеся к выполнению задач, могут быть настроены заранее в коде. Настройки в Flower предназначены, скорее, для чего-то срочного и для того, чтобы даже тестировщик или ПМ, зашедший в Flower, мог «погасить пожар», который там может произойти.

  Заключение

  Сегодня мы познакомились с базовыми возможностями Celery и Flower. Celery позволяет связывать задачи между собой, создавать очереди и переносить таски в фоновую обработку. А при помощи веб-инструмента Flower вы будете в курсе выполнения задач в реальном времени, сможете отслеживать и перезапускать воркеры.

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

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

  1. Как открыть окна каскадом
  2. Как приподнять стол
  3. Как создать пдф файл на компьютере
  4. Какие агрегатные функции можно использовать в операторе select