셀러리는 배우기가 어려울 수 있습니다. 문서는 포괄적이지만 기본 사항을 건너뛰는 경향이 있습니다.

이 게시물에서는 Celery의 네 가지 주요 개념을 정의하고, Celery와 Kombu의 관계에 대해 논의하고, 몇 가지 코드 예제를 사용하여 Celery가 실제 애플리케이션에서 어떻게 유용할 수 있는지 보여줍니다. 예제에서는 Django 웹 프레임워크와 해당 @shared_task 데코레이터를 사용하지만 개념은 Flask, FastAPI 및 기타에도 적용 가능합니다.

작업, 브로커, 작업자 및 백엔드

현재 Celery 문서에서 브로커 또는 백엔드로 간주되는 내용을 명확하게 설명하는 위치를 찾기가 어려울 것입니다. 정의를 추론합니다.

다음은 Celery를 시작하기 전에 알아야 할 개념입니다.

일

작업은 Celery가 비동기적으로 수행하는 작업입니다(이 맥락에서는 "즉각적이지 않음"을 뜻하는 멋진 단어입니다). 웹 애플리케이션에서는 사용자가 양식을 제출한 후 이메일을 보내는 작업이 있을 수 있습니다. 이메일 전송은 몇 초의 작업이 소요될 수 있으며 리디렉션하기 전에 사용자가 이메일이 전송될 때까지 기다리도록 하면 애플리케이션이 느리게 느껴질 수 있습니다.

작업은 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.",
        "[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에서 데코레이터를 사용하여 작업을 정의하면 작업에 지연() 메서드가 추가됩니다. 양식이 성공적으로 저장된 후 위의 예에서 Delay() 메서드를 호출하는 send_thank_you_email 작업을 볼 수 있습니다. Delay()가 호출되면 send_thank_you_email 작업과 해당 데이터가 저장된 브로커로 전송되고 나중에 worker에 의해 실행됩니다. 이메일로 보내드립니다.

양식을 저장한 후 추가 이메일을 보내야 하는 경우 작업을 Celery로 푸시하면 이점이 더욱 분명해집니다. 예를 들어 고객 지원팀에 새로운 피드백을 받았다는 이메일을 보낼 수 있습니다. Celery를 사용하면 응답에 추가 시간이 거의 추가되지 않습니다.

셀러리 작업을 통해 추가 고급 구성도 가능합니다. 이메일 전송에 실패한 경우 자동으로 재시도하고 max_retries, retry_backoff, retry_jitter 등과 같은 설정을 구성하도록 작업을 코딩할 수 있습니다.

브로커

셀러리 향상 제안 용어집에는 메시지 브로커:

엔터프라이즈 통합 패턴은 메시지 브로커를 여러 대상에서 메시지를 수신하고 올바른 대상을 결정하며 메시지를 올바른 채널로 라우팅할 수 있는 아키텍처 빌딩 블록으로 정의합니다.

Celery의 목적을 위해 브로커를 생성된 작업이 저장되는 "메시지 전송"으로 간주합니다. 브로커는 실제로 작업을 실행하지 않습니다. 이는 작업자의 작업입니다. 대신 브로커는 작업이 예약될 때 예약된 작업을 에 저장하고 작업자가 최종적으로 작업을 실행할 때 에서 가져오는 장소입니다. 브로커는 Celery가 작동하는 데 필수 구성 요소이며 Celery는 정확히 하나의 브로커에 연결됩니다.

Celery의 백엔드 및 브로커 페이지에는 지원되는 브로커 중 일부가 나열되어 있으며, 나열되지 않았지만 지원하는 다른 실험적인 브로커(예: SQLAlchemy)가 있습니다. 이러한 브로커(또는 "메시지 전송")는 Kombu라는 메시지 전송을 위해 Celery에서 유지 관리하는 Python 라이브러리에 의해 관리됩니다. 브로커 구성에 대한 정보를 찾을 때 Celery 문서보다는 Kombu 문서를 참조하는 것이 도움이 되는 경우가 있습니다.

일부 브로커에는 작업 팬아웃 및 우선순위와 같은 고급 기능이 있는 반면 다른 브로커는 간단한 대기열로 작동합니다.

노동자

worker는 브로커에서 작업을 가져오고 Python 앱에 정의된 작업 기능을 실행하는 Celery 인스턴스입니다. Celery 자체가 Python으로 작성되었기 때문에 Celery는 워커에서 Python 코드를 실행할 수 있습니다.

많은 작업자가 동시에 실행하여 작업을 실행할 수 있습니다. 셀러리 작업자 명령을 실행하면 기본적으로 컴퓨터의 모든 코어에 대해 작업자가 실행됩니다. 컴퓨터의 코어가 16개인 경우 Celery Worker를 실행하면 16개의 작업자가 시작됩니다.

실행 중인 작업자가 없으면 작업자가 이를 실행할 수 있을 때까지 메시지(작업)가 브로커에 누적됩니다.

백엔드

Celery 사용자 가이드의 작업 페이지에는 백엔드:

작업을 추적하고 싶거나 반환 값이 필요한 경우 Celery는 나중에 검색할 수 있도록 상태를 어딘가에 저장하거나 보내야 합니다. SQLAlchemy/Django ORM, Memcached, RabbitMQ/QPid(rpc) 및 Redis 중에서 선택할 수 있는 여러 내장 결과 백엔드가 있습니다. 또는 직접 정의할 수도 있습니다.

TLDR: 백엔드는 비동기 작업의 결과와 반환된 결과를 추적합니다. 이는 실제로 무엇을 의미하며, 언제 유용할 수 있나요?

연례 보고서를 생성할 수 있는 회계 앱을 Django에서 구축한다고 상상해 보세요. 보고서를 생성하는 데 몇 분 정도 걸릴 수 있습니다.

사용자에게 보다 응답성이 뛰어난 환경을 제공하기 위해 AJAX 요청을 사용하여 보고서 생성 작업을 시작합니다. 해당 요청은 보고서가 생성되었는지 확인하기 위해 몇 초마다 서버를 폴링하는 데 사용할 수 있는 작업 ID를 반환합니다. 작업이 완료되면 클라이언트가 JavaScript를 통해 보고서에 대한 링크를 표시하는 데 사용할 수 있는 보고서 ID가 반환됩니다.

다음 코드를 사용하여 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()에 의해 반환된 자산 ID는 백엔드에 저장됩니다. 백엔드는 결과와 반환된 결과를 저장합니다. 백엔드는 아직 처리되지 않은 작업의 상태를 저장하지 않습니다. 이는 결과가 나온 후에만 추가됩니다. "PENDING"을 반환하는 작업은 완전히 알 수 없는 상태입니다. 연결된 작업이 존재하지 않을 수도 있습니다. 작업은 일반적으로 "SUCCESS" 또는 "FAILURE"를 반환하지만 Celery 상태 문서에서 모든 상태를 볼 수 있습니다.

Celery가 작업을 실행하는 데 백엔드가 반드시 필요한 것은 아닙니다. 그러나 작업 결과를 확인하거나 작업 결과를 반환해야 하는 경우에는 백엔드가 필요합니다. Celery에 백엔드가 구성되어 있지 않은 경우 작업 상태를 확인하려고 하면 예외가 발생합니다.

이 게시물이 Celery의 개별 부분과 Celery 사용을 고려하는 이유를 이해하는 데 도움이 되기를 바랍니다. 공식 문서는 이해하기 어렵지만 Celery를 깊이 배우면 Python 애플리케이션 내에서 새로운 가능성을 열어줄 수 있습니다.