Celery для новичков
Привет, Хабр!
Celery — это асинхронная распределенная очередь задач, написанная на Python, она предназначена для обработки сообщений в реальном времени при помощи многозадачности. Используя Celery, можно организовать выполнение задач в фоновом режиме, не загружая основной поток приложения.
Используя Celery можно легко организовать выполнение фоновых задач.
Установка осуществляется через pip.
Основные возможности Celery
Определение задач
Создадим экземпляр Celery в файле celery_app.py:
from celery import Celery
app = Celery('example', broker='your_broker_url_here')Можно определить фоновые задачи с помощью декоратора @app.task. К примеру функция, которая просто складывает два числа:
@app.task
def add(x, y):
return x + yadd — это асинхронная задача. Можно вызвать её с помощью add.delay(x, y).
Celery предлагает параметры для настройки задач:
ignore_result
Если не нужен результат выполнения задачи, есть параметр ignore_result:
@app.task(ignore_result=True)
def add(x, y):
return x + yrate_limit ограничивает скорость выполнения задач. Например, если нужно, чтобы задача add выполнялась чаще, чем 10 раз в минуту, можно настроить rate_limit:
@app.task(rate_limit='10/m')
def add(x, y):
return x + yretry
Задача может иногда терпеть неудачу из-за проблем (которые вечно бывают):
@app.task(bind=True, max_retries=3, default_retry_delay=60)
def add(self, x, y):
try:
# Попытка выполнения задачи
return x + y
except SomeTemporaryException as exc:
# Запланировать повторное выполнение задачи
raise self.retry(exc=exc)
Задача попытается выполниться до 3 раз с интервалом в 60 секунд, если возникнет временная ошибка.
Вызов задач
Метод delay() — это самый простой способ вызвать задачу асинхронно. Под капотом он использует apply_async():
from tasks import add
# Вызов задачи асинхронно
result = add.delay(4, 4)apply_async() предлагает больше гибкости, позволяя указать различные параметры выполнения: время и приоритет выполнения, а также колбэки и errbacks:
result = add.apply_async((4, 4), countdown=10)add будет запланирована к выполнению через 10 секунд после вызова.
signature() создает подпись задачи, которую можно использовать для создания сложных раб. процессов:
from celery import signature
sig = signature('tasks.add', args=(2, 2), immutable=True)
sig.delay()chain() позволяет соединить несколько задач в одну последовательность, где результат одной задачи передается в качестве аргумента следующей:
from celery import chain
# (4 + 4) -> (8 * 10)
res = chain(add.s(4, 4), multiply.s(10))()group() используется для параллельного выполнения набора задач. Он возвращает специальный объект GroupResult, который позволяет отслеживать выполнение группы задач:
from celery import group
# выполняет add(2, 2) и add(4, 4) параллельно
group_result = group(add.s(2, 2), add.s(4, 4))()chord() — это комбинация group() и chain(), позволяющая выполнить группу задач параллельно и затем вызвать callback-задачу с результатами группы:
from celery import chord
# cначала выполняет add(2, 2) и add(4, 4) параллельно, затем результаты передаются в multiply()
result = chord([add.s(2, 2), add.s(4, 4)])(multiply.s(2))Различные примеры применения
Масштабируемая система обработки изображений
from celery import Celery
import PIL
from PIL import Image
app = Celery('image_processor', broker='pyamqp://guest@localhost//')
@app.task
def resize_image(image_path, output_path, size):
with Image.open(image_path) as img:
img.thumbnail(size, PIL.Image.ANTIALIAS)
img.save(output_path)
@app.task
def crop_image(image_path, output_path, crop_box):
with Image.open(image_path) as img:
cropped_img = img.crop(crop_box)
cropped_img.save(output_path)
# юзаем
resize_image.delay('path/to/image.jpg', 'path/to/resized_image.jpg', (800, 600))
crop_image.delay('path/to/image.jpg', 'path/to/cropped_image.jpg', (100, 100, 400, 400))Асинхронная рассылка уведомлений
Celery часто юзают для асинхронной рассылки электронных писем, SMS, или push-уведомлений пользователям. Пример отправки электронных писем с помощью SMTP:
from celery import Celery
import smtplib
app = Celery('notifications', broker='pyamqp://guest@localhost//')
@app.task
def send_email(recipient, subject, body):
server = smtplib.SMTP('smtp.example.com', 587)
server.starttls()
server.login("your_email@example.com", "your_password")
message = f"Subject: {subject}\n\n{body}"
server.sendmail("your_email@example.com", recipient, message)
server.quit()
# пример использования
send_email.delay("user@example.com", "Welcome!", "Thank you for registering with us.")Асинхронное выполнение длительных задач на примере генерации отчетов
from celery import Celery
import time
app = Celery('reports', broker='pyamqp://guest@localhost//')
@app.task
def generate_report(report_id, parameters):
# операции по извлечению данных, их обработке и анализе
time.sleep(60) # имитация длительной операции
# сохранение или отправка отчета
return f"Report {report_id} generated with parameters {parameters}"
# использование
generate_report.delay("report_123", {"param1": "value1", "param2": "value2"})Интеграция с Django
Cоздаем проект django и добавляем приложение:
django-admin startproject myproject
cd myproject
django-admin startapp myappСоздаем файл celery.py в корневой директории проекта Django (на уровне settings.py):
# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()Добавляем следующие строкиsettings.py, чтобы указать Celery использовать Redis в качестве брокера сообщений:
# myproject/settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'Переходим в приложение, созданное внутри проекта Django, и создайте файл tasks.py. Например, создадим задачу для сложения двух чисел:
# myapp/tasks.py
from celery import shared_task
@shared_task
def add(x, y):
return x + yТеперь можно юзать эту задачу асинхронно из моделей или любой другой части Django. Например, добавим вызов задачи:
# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()Для выполнения асинхронных задач необходимо запускаем Celery worker:
celery -A myproject worker --loglevel=infoЭто запустит worker, который будет слушать и выполнять асинхронные задачи.
Можно еще подрубить мониторинг задач и воркеров с flower:
celery -A myproject flowerНапомню, что в рамках онлайн-курсов OTUS вы можете изучить самые популярные ЯП, а также зарегистрироваться на ряд бесплатных мероприятий.
