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 + y

add — это асинхронная задача. Можно вызвать её с помощью add.delay(x, y).

Celery предлагает параметры для настройки задач:

ignore_result

Если не нужен результат выполнения задачи, есть параметр ignore_result:

@app.task(ignore_result=True)
def add(x, y):
    return x + y

rate_limit ограничивает скорость выполнения задач. Например, если нужно, чтобы задача add выполнялась чаще, чем 10 раз в минуту, можно настроить rate_limit:

@app.task(rate_limit='10/m')
def add(x, y):
    return x + y

retry

Задача может иногда терпеть неудачу из-за проблем (которые вечно бывают):

@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 вы можете изучить самые популярные ЯП, а также зарегистрироваться на ряд бесплатных мероприятий.