Celery のタスク、ブローカー、ワーカー、バックエンドについて理解する

セロリを学ぶのは気が遠くなるかもしれません。ドキュメントは包括的ですが、基本的な部分を省略する傾向があります。

この投稿では、Celery の 4 つの主要な概念を定義し、Celery と Kombu の関係について説明し、いくつかのコード例を使用して Celery が実際のアプリケーションでどのように役立つかを説明します。この例では、Django Web フレームワークとその @shared_task デコレーターを使用しますが、この概念は Flask、FastAPI などにも適用できます。

タスク、ブローカー、ワーカー、バックエンド

現在の Celery ドキュメントで、ブローカー または バックエンド とみなされるものを明確に説明している場所を見つけるのは難しいでしょう。しかし、十分に掘れば、定義を推測します。

以下は、Celery を始める前に知っておくべき概念です。

タスク

タスクは、Celery が 非同期 で実行する作業です (この文脈では、「すぐにではない」という意味の派手な言葉です)。 Web アプリケーションでは、ユーザーがフォームを送信した後に電子メールを送信するタスクが考えられます。電子メールの送信には数秒かかる操作があり、リダイレクトする前に電子メールの送信をユーザーに待たせると、アプリケーションが遅く感じる可能性があります。

タスクは Celery のデコレータを使用して定義されます。以下では、@shared_task デコレーターを使用して、send_thank_you_email() を submit_feedback() フォーム送信ハンドラーで使用できる Celery タスクに変換します。

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.",
        "noreply@example.com",
        [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 のデコレータを使用してタスクが定義されると、タスクに late() メソッドが追加されます。フォームが正常に保存された後、上の例では send_thank_you_email タスクが late() メソッドを呼び出していることがわかります。 late() が呼び出されると、send_Thank_you_email タスクとそのデータが ブローカー に送信され、そこに保存され、後で ワーカー によって実行されます。その時点でユーザーはメールで送信されます。

フォームを保存した後に追加の電子メールを送信する必要がある場合、作業を Celery にプッシュする利点がより明らかになります。たとえば、新しいフィードバックを受け取ったことをカスタマー サポート チームに電子メールで送信することができます。 Celery を使用すると、応答に追加の時間がほとんど追加されません。

Celery タスクでは、追加の高度な構成も可能です。電子メールの送信に失敗した場合は、タスクをコーディングして自動的に再試行し、max_retries、retry_backoff、retry_jitter などの設定を構成できます。

ブローカ

Celery Enhancement Proposals の用語集には、メッセージ ブローカー について次のように記載されています。

エンタープライズ統合パターンでは、メッセージ ブローカーを、複数の宛先からメッセージを受信し、正しい宛先を決定し、正しいチャネルにメッセージをルーティングできるアーキテクチャのビルディング ブロックとして定義します。

Celery の目的のために、ブローカー を、作成されたタスクが保存される「メッセージ トランスポート」とみなします。ブローカーは実際にタスクを実行しません。それはワーカーの仕事です。代わりに、ブローカーは、タスクがスケジュールされたときにスケジュールされたタスクが保存され、ワーカーが最終的にタスクを実行するときにからプルされる場所です。ブローカーは Celery が動作するために必須のコンポーネントであり、Celery は 1 つのブローカーにのみ接続します。

Celery のバックエンドとブローカーのページには、サポートされているブローカーの一部がリストされています。また、リストされていない、サポートされている実験的なブローカーもあります (SQLAlchemy など)。これらのブローカー (または「メッセージ トランスポート」) は、Celery が管理する Kombu と呼ばれるメッセージ トランスポート用の Python ライブラリによって管理されます。ブローカーの構成に関する情報を探す場合、Celery のドキュメントではなく Kombu のドキュメントを参照すると役立つ場合があります。

一部のブローカーはタスクのファンアウトや優先度などの高度な機能を備えていますが、他のブローカーは単純なキューとして動作します。

ワーカー

ワーカー は、ブローカーからタスクを取得し、Python アプリで定義されたタスク関数を実行する Celery のインスタンスです。 Celery 自体は Python で書かれているため、Celery はワーカー内で Python コードを実行できます。

多くのワーカーを同時に実行してタスクを実行できます。 celery ワーカー コマンドを実行すると、デフォルトでコンピューターのすべてのコアに対してワーカーが起動されます。コンピューターに 16 コアがある場合、celery ワーカーを実行すると 16 個のワーカーが開始されます。

実行中のワーカーがない場合、メッセージ (タスク) は、ワーカーが実行できるようになるまでブローカーに蓄積されます。

バックエンド

Celery ユーザー ガイドのタスク ページには、バックエンド について次のように記載されています。

If you want to keep track of tasks or need the return values, then Celery must store or send the states somewhere so that they can be retrieved later. There are several built-in result backends to choose from: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc), and Redis – or you can define your own.

TLDR: a backend tracks the outcomes and returned results of async tasks. What does that actually mean, and when could it be useful?

Imagine you are building an accounting app in Django that can generate an annual report. The report could take minutes to generate.

To give your users a more responsive experience, you use an AJAX request to kick off a report generation task. That request returns an ID of the task, which it can use to poll the server every few seconds to see if the report is generated. Once the task is complete, it will return the ID of the report, which the client can use to display a link to the report via JavaScript.

We can implement this with Celery and Django using the following code:

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})

In this example, the asset ID returned by generate_report_task() is stored in a backend. The backend stores the outcome and returned result. The backend does not store the status of yet-to-be-processed tasks: these will only be added once there has been an outcome. A task that returns "PENDING" has a completely unknown status: an associated task might not even exist. Tasks will typically return "SUCCESS" or "FAILURE", but you can see all statuses in the Celery status docs.

Having a backend is not required for Celery to run tasks. However, it is required if you ever need to check the outcome of a task or return a task's result. If you try to check a task's status when Celery does not have a backend configured, an exception will be raised.

---

I hope this post helps you understand the individual pieces of Celery and why you might consider using it. While the official documentation is challenging to grok, learning Celery deeply can unlock new possibilities within your Python applications.

以上がCelery のタスク、ブローカー、ワーカー、バックエンドについて理解するの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。