Es kann entmutigend sein, Sellerie zu lernen. Obwohl die Dokumentation umfassend ist, neigt sie dazu, die Grundlagen zu überspringen.

In diesem Beitrag werden vier der Hauptkonzepte in Celery definiert, die Beziehung zwischen Celery und Kombu erörtert und anhand einiger Codebeispiele veranschaulicht, wie Celery in realen Anwendungen nützlich sein könnte. In den Beispielen werden das Django-Webframework und sein @shared_task-Dekorator verwendet, aber die Konzepte sind auch auf Flask, FastAPI und andere anwendbar.

Aufgaben, Broker, Worker und Backends

Es wird Ihnen schwer fallen, in der aktuellen Celery-Dokumentation eine Stelle zu finden, die klar darlegt, was ein Broker oder ein Backend ist, aber mit genügend Recherche können Sie finden und Definitionen ableiten.

Im Folgenden finden Sie Konzepte, die Sie kennen sollten, bevor Sie mit Sellerie beginnen.

Aufgabe

Eine Aufgabe ist eine Arbeit, die Celery asynchron ausführt (in diesem Zusammenhang ist das ein schickes Wort für „nicht sofort“). In einer Webanwendung könnte eine Aufgabe darin bestehen, eine E-Mail zu senden, nachdem ein Benutzer ein Formular übermittelt hat. Das Senden einer E-Mail kann ein Vorgang von mehreren Sekunden sein, und wenn ein Benutzer gezwungen wird, auf das Senden einer E-Mail zu warten, bevor er umgeleitet wird, kann dies dazu führen, dass sich eine Anwendung langsam anfühlt.

Aufgaben werden mithilfe von Dekoratoren in Celery definiert. Im Folgenden verwenden wir den @shared_task-Dekorator, um send_thank_you_email() in eine Celery-Aufgabe umzuwandeln, die im Submit_feedback()-Formularübermittlungshandler verwendet werden kann.

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

Wenn eine Aufgabe mithilfe eines Dekorators in Celery definiert wird, wird der Aufgabe eine Methode „delay()“ hinzugefügt. Im obigen Beispiel können Sie sehen, wie die Aufgabe send_thank_you_email die Methode „delay()“ aufruft, nachdem das Formular erfolgreich gespeichert wurde. Wenn „delay()“ aufgerufen wird, sendet es die Aufgabe „send_thank_you_email“ und ihre Daten an den Broker, wo sie gespeichert werden und später von einem Worker ausgeführt werden, woraufhin der Benutzer dies tut per E-Mail verschickt werden.

Der Vorteil der Verlagerung der Arbeit auf Celery wird deutlicher, wenn Sie nach dem Speichern des Formulars weitere E-Mails senden müssen. Beispielsweise möchten Sie dem Kundensupport-Team möglicherweise per E-Mail mitteilen, dass es neues Feedback erhalten hat. Bei Sellerie bedeutet dies fast keine zusätzliche Reaktionszeit.

Sellerie-Aufgaben ermöglichen auch eine zusätzliche erweiterte Konfiguration. Für den Fall, dass eine E-Mail nicht gesendet werden konnte, können Sie Ihre Aufgabe so programmieren, dass sie automatisch erneut versucht, und Einstellungen wie max_retries, retry_backoff, retry_jitter usw. konfigurieren.

Makler

Das Glossar der Celery Enhancement Proposals sagt Folgendes über Message Brokers:

Enterprise Integration Patterns definiert einen Message Broker als einen Architekturbaustein, der Nachrichten von mehreren Zielen empfangen, das richtige Ziel bestimmen und die Nachricht an den richtigen Kanal weiterleiten kann.

Für unsere Zwecke mit Celery betrachten wir einen Broker als „Nachrichtentransport“, in dem erstellte Aufgaben gespeichert werden. Makler führen die Aufgabe nicht wirklich aus: Das ist die Aufgabe eines Arbeiters. Stattdessen sind Broker der Ort, an dem geplante Aufgaben gespeichert werden, wenn eine Aufgabe geplant wird, und von abgerufen werden, wenn ein Mitarbeiter schließlich eine Aufgabe ausführt. Ein Broker ist eine erforderliche Komponente, damit Celery funktioniert, und Celery stellt eine Verbindung zu genau einem Broker her.

Die Seite „Backends und Broker“ von Celery listet einige der unterstützten Broker auf, und es gibt andere unterstützte experimentelle Broker, die nicht aufgeführt sind (z. B. SQLAlchemy). Diese Broker (oder „Nachrichtentransporte“) werden von einer von Celery gepflegten Python-Bibliothek für Nachrichtentransporte namens Kombu verwaltet. Wenn Sie nach Informationen zum Konfigurieren von Brokern suchen, ist es manchmal hilfreich, die Dokumentation von Kombu anstelle der von Celery zu konsultieren.

Einige Broker verfügen über erweiterte Funktionen wie Task-Fanout und Priorität, während andere als einfache Warteschlangen arbeiten.

Arbeiter

Ein

Worker ist eine Instanz von Celery, die Aufgaben vom Broker abruft und die in Ihrer Python-App definierten Aufgabenfunktionen ausführt. Celery kann Python-Code in seinen Workern ausführen, da Celery selbst in Python geschrieben ist.

Viele Arbeiter können gleichzeitig laufen, um Aufgaben auszuführen. Wenn Sie den Befehl celery worker ausführen, wird standardmäßig ein Worker für jeden Kern Ihres Computers gestartet. Wenn Ihr Computer über 16 Kerne verfügt, werden beim Ausführen von celery worker 16 Worker gestartet.

Wenn keine Worker ausgeführt werden, sammeln sich Nachrichten (Aufgaben) im Broker an, bis Worker zur Ausführung verfügbar sind.

Backend

Die Aufgabenseite im Celery-Benutzerhandbuch enthält Folgendes zu

Backends:

Wenn Sie den Überblick über Aufgaben behalten möchten oder die Rückgabewerte benötigen, muss Celery die Zustände irgendwo speichern oder senden, damit sie später abgerufen werden können. Es stehen mehrere integrierte Ergebnis-Backends zur Auswahl: SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid (rpc) und Redis – oder Sie können Ihr eigenes definieren.

TLDR: Ein Backend verfolgt die Ergebnisse und zurückgegebene Ergebnisse von asynchronen Aufgaben. Was bedeutet das eigentlich und wann könnte es nützlich sein?

Stellen Sie sich vor, Sie erstellen eine Buchhaltungs-App in Django, die einen Jahresbericht erstellen kann. Die Erstellung des Berichts kann Minuten dauern.

Um Ihren Benutzern ein reaktionsschnelleres Erlebnis zu bieten, verwenden Sie eine AJAX-Anfrage, um eine Aufgabe zur Berichterstellung zu starten. Diese Anfrage gibt eine ID der Aufgabe zurück, mit der sie alle paar Sekunden den Server abfragen kann, um zu sehen, ob der Bericht generiert wurde. Sobald die Aufgabe abgeschlossen ist, wird die ID des Berichts zurückgegeben, mit der der Client über JavaScript einen Link zum Bericht anzeigen kann.

Wir können dies mit Celery und Django mithilfe des folgenden Codes implementieren:

from celery import shared_task Von django.http JsonResponse importieren aus django.views.decorators.http import require_http_methods aus Accounting.models Import Asset aus Accounting.reports import AnnualReportGenerator @shared_task def generic_report_task(year): # Dies kann Minuten dauern... report = AnnualReportGenerator().generate(year) asset = Asset.objects.create( name=f"{year} Jahresbericht", url=report.url, ) Geben Sie asset.id zurück @require_http_methods(["POST"]) def generic_annual_report_view(request): Jahr = request.POST.get("Jahr") task = generic_report_task.delay(Jahr) JsonResponse({"taskId": task.id}) zurückgeben def get_annual_report_generation_status_view(request, task_id): task = generic_report_task.AsyncResult(task_id) # Der Status ist typischerweise „PENDING“, „SUCCESS“ oder „FAILURE“ status = task.status return JsonResponse({"status": status, "assetId": task.result})

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 diesem Beispiel wird die von generic_report_task() zurückgegebene Asset-ID in einem

Backend gespeichert. Das Backend speichert das Ergebnis und das zurückgegebene Ergebnis. Das Backend speichert nicht den Status noch zu bearbeitender Aufgaben: Diese werden erst hinzugefügt, wenn ein Ergebnis vorliegt. Eine Aufgabe, die „PENDING“ zurückgibt, hat einen völlig unbekannten Status: Eine zugehörige Aufgabe existiert möglicherweise nicht einmal. Aufgaben geben normalerweise „ERFOLGREICH“ oder „FEHLER“ zurück, aber Sie können alle Status in den Celery-Statusdokumenten sehen.

Es ist nicht erforderlich, über ein Backend zu verfügen, damit Celery Aufgaben ausführen kann. Es ist jedoch erforderlich, wenn Sie jemals das Ergebnis einer Aufgabe überprüfen oder das Ergebnis einer Aufgabe zurückgeben müssen. Wenn Sie versuchen, den Status einer Aufgabe zu überprüfen, während für Celery kein Backend konfiguriert ist, wird eine Ausnahme ausgelöst.

---

Ich hoffe, dieser Beitrag hilft Ihnen, die einzelnen Selleriestücke zu verstehen und zu verstehen, warum Sie die Verwendung in Betracht ziehen könnten. Auch wenn es schwierig ist, sich mit der offiziellen Dokumentation vertraut zu machen, kann das gründliche Erlernen von Celery neue Möglichkeiten in Ihren Python-Anwendungen eröffnen.