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

В этом посте будут определены четыре основные концепции Celery, обсуждена связь между Celery и Kombu, а также использовано несколько примеров кода, чтобы проиллюстрировать, как Celery может быть полезен в реальных приложениях. В примерах будет использоваться веб-фреймворк Django и его декоратор @shared_task, но эти концепции также применимы к Flask, FastAPI и другим.

Задачи, брокеры, воркеры и бэкенды

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

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

Задача

задача — это некоторая часть работы, которую Celery выполнит асинхронно (в данном контексте это красивое слово, означающее «не сразу»). В веб-приложении одной задачей может быть отправка электронного письма после отправки пользователем формы. Отправка электронного письма может занять несколько секунд, и принуждение пользователя ждать отправки электронного письма перед перенаправлением может привести к замедлению работы приложения.

Задачи определяются с помощью декораторов в Celery. Ниже мы используем декоратор @shared_task, чтобы превратить send_thank_you_email() в задачу Celery, которую можно использовать в обработчике отправки формы submit_feedback().

from config.celery import shared_task
from django.core.mail import send_mail
from django.shortcuts import render, redirect
from feedback.forms import FeedbackForm

@shared_task
def send_thank_you_email(email_address):
    send_mail(
        "Thank you for your feedback!",
        "We appreciate your input.",
        "[email protected]",
        [email_address],
    )

def submit_feedback(request):
    if request.method == "POST":
        form = FeedbackForm(request.POST)
        if form.is_valid():
            form.save()

            # Push the task to the broker using the delay() method.
            send_thank_you_email.delay(form.cleaned_data["email"])

            return redirect("/thank-you/")
    else:
        form = FeedbackForm()

    return render(request, "feedback.html", {"form": form})

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

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

Задачи Celery также допускают дополнительную расширенную настройку. В случае, если электронное письмо не удалось отправить, вы можете запрограммировать свою задачу для автоматического повтора и настроить такие параметры, как max_retries, retry_backoff, retry_jitter и т. д.

Маклер

В глоссарии предложений по улучшению сельдерея говорится следующее о брокерах сообщений:

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

Для наших целей с Celery мы будем рассматривать брокера как «транспорт сообщений», в котором хранятся созданные задачи. Брокеры на самом деле не выполняют задачу: это работа работника. Вместо этого брокеры — это место, где запланированные задачи сохраняются в, когда задача запланирована, и извлекаются из, когда работник в конечном итоге выполняет задачу. Брокер — это необходимый компонент для работы Celery, и Celery будет подключаться только к одному брокеру.

На странице «Бэкэнды и брокеры» Celery перечислены некоторые поддерживаемые им брокеры, а также есть другие поддерживаемые им экспериментальные брокеры, которые не указаны (например, SQLAlchemy). Эти брокеры (или «транспорты сообщений») управляются поддерживаемой Celery библиотекой Python для транспорта сообщений под названием Kombu. При поиске информации о настройке брокеров иногда полезно обратиться к документации Kombu, а не к документации Celery.

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

Рабочий

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

Многие работники могут выполнять задачи одновременно. Когда вы запускаете команду celery worker, она по умолчанию запускает работника для каждого ядра вашего компьютера. Если ваш компьютер имеет 16 ядер, при запуске рабочего процесса сельдерея запустится 16 рабочих процессов.

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

Бэкэнд

На странице задач в Руководстве пользователя Celery о бэкэндах говорится следующее:

Если вы хотите отслеживать задачи или вам нужны возвращаемые значения, тогда Celery должен хранить или отправлять куда-то состояния, чтобы их можно было получить позже. На выбор имеется несколько встроенных механизмов обработки результатов: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc) и Redis – или вы можете определить свой собственный.

TLDR: бэкэнд отслеживает результаты и возвращенные результаты асинхронных задач. Что это на самом деле означает и когда это может быть полезно?

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

Чтобы обеспечить более оперативное реагирование пользователей, вы используете запрос AJAX для запуска задачи создания отчета. Этот запрос возвращает идентификатор задачи, который можно использовать для опроса сервера каждые несколько секунд, чтобы проверить, сгенерирован ли отчет. После завершения задачи она вернет идентификатор отчета, который клиент может использовать для отображения ссылки на отчет через JavaScript.

Мы можем реализовать это с помощью Celery и Django, используя следующий код:

from celery import shared_task
from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from accounting.models import Asset
from accounting.reports import AnnualReportGenerator

@shared_task
def generate_report_task(year):
    # This could take minutes...
    report = AnnualReportGenerator().generate(year)
    asset = Asset.objects.create(
        name=f"{year} annual report",
        url=report.url,
    )
    return asset.id

@require_http_methods(["POST"])
def generate_annual_report_view(request):
    year = request.POST.get("year")
    task = generate_report_task.delay(year)
    return JsonResponse({"taskId": task.id})

def get_annual_report_generation_status_view(request, task_id):
    task = generate_report_task.AsyncResult(task_id)

    # The status is typically "PENDING", "SUCCESS", or "FAILURE"
    status = task.status
    return JsonResponse({"status": status, "assetId": task.result})

В этом примере идентификатор актива, возвращаемый функциейgenerate_report_task(), хранится в бэкэнде. Бэкэнд хранит результат и возвращенный результат. Бэкэнд не сохраняет статус еще не обработанных задач: они будут добавлены только после получения результата. Задача, возвращающая «ОЖИДАНИЕ», имеет совершенно неизвестный статус: связанная с ней задача может даже не существовать. Задачи обычно возвращают «УСПЕХ» или «НЕУДАЧА», но вы можете увидеть все статусы в документах о статусе Celery.

Для запуска задач Celery не требуется наличие бэкэнда. Однако он необходим, если вам когда-либо понадобится проверить результат задачи или вернуть результат задачи. Если вы попытаетесь проверить статус задачи, когда в Celery не настроен бэкэнд, будет выдано исключение.

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