Hallo ontwikkelaars! Zucht je ook al tijdens het zien van webdiensten die haperen door zware taken? Wanneer een gebruikersverzoek verschillende databases moet opvragen, externe API's moet aanroepen en zelfs beeldverwerking moet uitvoeren, dan is het geen verrassing dat onze diensten vaak vastlopen. Op zo'n moment verschijnt Celery als een redder in nood.

Velen van jullie hebben Celery gebruikt om achtergrondtaken te verwerken en de responstijd van de dienst te verbeteren. Met één regel, some_long_running_task.delay(args), wordt de zware taak als magie naar de achtergrond verplaatst en uitgevoerd, zodat de hoofddraad vrij is. Maar misschien heb je nog niet diepgaand onderzocht wat de echte aard en werking van delay() is. Vandaag gaan we de buitenkant en binnenkant van de delay() methode in detail bekijken, en je Celery-vaardigheden naar een hoger niveau tillen.

Celery – Snelheid bij het uitvoeren van asynchrone taken

1. Waarom hebben we Celery nodig?

Laten we kort opnieuw bekijken waarom Celery nodig is. Bij de ontwikkeling van webapplicaties worden we vaak geconfronteerd met de volgende scenario's.

  • Tijdrovende taken: E-mail verzenden, afbeeldingen/video's verwerken, complexe statistieken berekenen, massale gegevensimport/export, enz.
  • Afhankelijkheid van externe diensten: Vertraging bij het aanroepen van externe API's
  • Plotselinge pieken in verkeer: Een grote hoeveelheid verzoeken in een korte tijd kan de webserver overbelasten

Als we deze taken direct uitvoeren in de hoofddraad die gebruikersverzoeken verwerkt, moeten gebruikers wachten totdat de taak is voltooid. Dit leidt tot verhoogde responstijden van de dienst en uiteindelijk slechtere gebruikerservaring. Zelfs als de serverbronnen beperkt zijn of als een taak te lang duurt, kan dit leiden tot time-outs of een servercrash.

Celery is een gedistribueerd taakqueue-systeem dat deze problemen oplost. Door de verzoeken die snel moeten worden beantwoord onmiddellijk af te handelen, en de tijdrovende taken naar Celery te sturen voor asynchrone verwerking op de achtergrond, kunnen we de responstijd en stabiliteit van de dienst verbeteren.

2. Wat is de delay() methode?

Wat is precies de rol van de delay() methode die we vaak gebruiken?

delay() is de eenvoudigste manier om een Celery-taak asynchroon uit te voeren.

Functies die je definieert met de @app.task of @shared_task decorateurs zijn niet langer simpele Python-functies. Ze worden door Celery verpakt in een speciale Task object, dat methoden zoals delay() en apply_async() bevat. Op het moment dat je de delay() methode aanroept, vormt je code niet langer het daadwerkelijke uitvoeringscommando voor de taak, maar stuurt het informatie (functie naam, argumenten, enz.) in de vorm van een bericht naar de Celery message broker.

3. De werking van delay(): de reis achter de magie

De werking van delay() – Visualisatie van de asynchrone workflow

Hoewel delay() zinvol lijkt te zijn in één regel om asynchrone taken te creëren, komt daar een systematisch werkproces van Celery aan te pas. Hier zijn de stappen die plaatsvinden bij het aanroepen van delay().

  1. Taak aanroep (delay() aanroep): In de hoofdapplicatiecode, zoals bij een Django-view, roep je my_task_function.delay(arg1, arg2) aan.

  2. Bericht generatie: De Celery-cliënt (de Django-applicatie) genereert een bericht dat aangeeft dat de taak my_task_function met argumenten arg1 en arg2 moet worden uitgevoerd. Dit bericht is geserialiseerd in een gestandaardiseerd JSON- of Pickle-formaat dat de taaknaam, argumenten (args, kwargs) en indien nodig andere metadata (zoals taak-ID) bevat.

  3. Bericht verzenden naar broker: Het gegenereerde bericht wordt verzonden naar de Celery message broker. Dit is een berichtqueue-systeem zoals Redis, RabbitMQ of Kafka. De message broker slaat dit bericht op in een specifieke queue.

  4. Berichtontvangst door worker: Celery workers zijn verbonden met de message broker en pollende of subscriben continu op specifieke queues. Wanneer er een nieuw bericht aankomt in de queue, ontvangt de worker dit.

  5. Taak deserialisatie en uitvoering: De worker deserialiseert het ontvangen bericht om de taaknaam en argumenten te extraheren. Vervolgens zoekt hij de bijbehorende taakfunctie (my_task_function) en voert deze onafhankelijk uit binnen zijn eigen proces of thread.

  6. Resultaat opslaan (optioneel): Nadat de taak is voltooid, kan het resultaat worden opgeslagen in Celery's result backend. Deze kan Redis, een database (zoals Django ORM) of S3 zijn. Dit resultaat kan later worden opgevraagd via een AsyncResult object.

  7. Antwoord van de view: Terwijl dit alles op de achtergrond plaatsvindt, retourneert de hoofdapplicatie (bijvoorbeeld de Django-view) onmiddellijk een antwoord aan de client zodra de taak succesvol aan de queue is toegevoegd. De webverzoeken worden niet langer langdurig geblokkeerd.

Dankzij deze gescheiden architectuur kan de webserver snel reageren op verzoeken, terwijl zware taken aan Celery workers worden gedelegeerd, wat de algehele prestaties en schaalbaarheid van het systeem aanzienlijk verbetert.

4. Voorbeelden van delay() gebruik

Laten we een paar van de meest voorkomende scenario's van het gebruik van delay() bekijken.

Voorbeeld 1: Eenvoudige e-mail verzendtaak

# myapp/tasks.py
from celery import shared_task
import time

@shared_task
def send_email_task(recipient_email, subject, message):
    print(f"E-mail verzenden naar {recipient_email} - Onderwerp: {subject}")
    time.sleep(5) # Tijdvertraag voor het simuleren van e-mailverzending
    print(f"E-mail verzonden naar {recipient_email}")
    return True

# myapp/views.py
from django.http import HttpResponse
from .tasks import send_email_task

def contact_view(request):
    if request.method == 'POST':
        recipient = request.POST.get('email')
        sub = "Bedankt voor uw vraag."
        msg = "Uw vraag is succesvol ontvangen."

        # E-mail verzendtaak asynchroon uitvoeren
        send_email_task.delay(recipient, sub, msg)

        return HttpResponse("Uw vraag is ontvangen en de e-mail zal binnenkort worden verzonden.")
    return HttpResponse("Dit is de contactpagina.")

Wanneer een gebruiker het contactformulier indient, wordt send_email_task.delay() aangeroepen en wordt de e-mail verzendtaak naar de achtergrond verplaatst. De webserver retourneert onmiddellijk een antwoord, zodat gebruikers niet hoeven te wachten tot de e-mail verzending is voltooid.

Voorbeeld 2: Afbeelding thumbnail genereren taak

# myapp/tasks.py
from celery import shared_task
import os
from PIL import Image # Pillow-bibliotheek vereist: pip install Pillow

@shared_task
def create_thumbnail_task(image_path, size=(128, 128)):
    try:
        img = Image.open(image_path)
        thumb_path = f"{os.path.splitext(image_path)[0]}_thumb{os.path.splitext(image_path)[1]}"
        img.thumbnail(size)
        img.save(thumb_path)
        print(f"Thumbnail aangemaakt voor {image_path} op {thumb_path}")
        return thumb_path
    except Exception as e:
        print(f"Fout bij het maken van thumbnail voor {image_path}: {e}")
        raise

# myapp/views.py
from django.http import HttpResponse
from .tasks import create_thumbnail_task

def upload_image_view(request):
    if request.method == 'POST' and request.FILES.get('image'):
        uploaded_image = request.FILES['image']
        # Afbeelding opslaan naar een tijdelijke locatie (gebruik S3 of andere opslag in een echte service)
        save_path = f"/tmp/{uploaded_image.name}"
        with open(save_path, 'wb+') as destination:
            for chunk in uploaded_image.chunks():
                destination.write(chunk)

        # Thumbnail genereren taak asynchroon uitvoeren
        create_thumbnail_task.delay(save_path)

        return HttpResponse("Afbeelding is geüpload en thumbnail generatie is aan de gang op de achtergrond.")
    return HttpResponse("Dit is de afbeelding uploadpagina.")

Zevenals tijdsintensievere werkzaamheden zoals het uploaden van afbeeldingen, kunnen deze ook asynchroon worden verwerkt door delay() toe te passen om de belasting op de webserver te verminderen.

5. Voordelen en beperkingen van delay()

Voordelen:

  • Begrijpelijke en intuïtieve API: Dit is de grootste troef. delay() is eenvoudig in gebruik, zodat je taken snel aan de queue kunt toevoegen zonder extra opties.
  • Verbeterde responsiviteit: Het voorkomt dat de webverzoekverwerkingsdraad wordt geblokkeerd door taken naar de achtergrond te sturen en gebruikers een snellere respons te bieden.
  • Schaalbaarheid: Je kunt werklasten eenvoudig verdelen en aantallen workers aanpassen om de doorvoer te regelen.
  • Betrouwbaarheid: Zelfs als een specifieke taak faalt, heeft dit geen impact op de webserver als geheel, en er kunnen mechanismen voor herhalingen worden gebruikt voor betrouwbare verwerking.

Beperkingen:

  • Ondersteuning voor eenvoudige opties: delay() is de meest basale manier om een taak aan de queue toe te voegen. Het biedt geen ondersteuning voor geavanceerde opties zoals het uitstellen van taken, sturen naar specifieke queues of prioriteiten instellen. Voor dat soort functionaliteiten moet je apply_async() gebruiken.
  • Eenvoud van foutafhandeling: De aanroep van delay() retourneert alleen of de taak succesvol in de queue is geplaatst. Je weet echter niet of de daadwerkelijke taak succesvol was of wat het resultaat was. Hiervoor moet je gebruik maken van de results backend en de AsyncResult objecten.

Conclusie

Vandaag hebben we uitgebreid gekeken naar de werking en toepassingen van de belangrijke delay() methode van Celery. Ik hoop dat delay() niet alleen een handige syntaxis is, maar ook de sleutel tot het begrijpen is van hoe het gedistribueerde taakqueue-systeem van Celery werkt.

In de volgende post zullen we dieper ingaan op de apply_async() methode, die een superieure variant van delay() kan zijn, en we zullen duidelijke richtlijnen geven over de relatie tussen beide methoden en wanneer ze moeten worden gebruikt. Blijf kijken!