Celery와 AWS SQS를 사용한 작업 라우팅 및 처리방법과 옵션 기능으로 기본개념 알아보기

import os
from celery import Celery
from kombu import Exchange, Queue

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.local")

# 이곳에서 기본적인 환경을 구성합니다.
app = Celery("config")
app.config_from_object("django.conf:settings", namespace="CELERY")

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="KST",
    enable_utc=True,
    result_backend=None,
    broker_url='sqs://<aws_access_key_id>:<aws_secret_access_key>@',
    broker_transport_options={
        'region': 'ap-northeast-2', # 각자의 환경으로 설정하면 됩니다.
        'polling_interval': 5,    # 메시지 폴링 간격
        'visibility_timeout': 60,  # 가시성 타임아웃
        'wait_time_seconds': 20,
        'queue_name_prefix': 'celery-',
    },
)

#SQS에서는 필요없음
app.conf.task_queues = (
    Queue("add_function", Exchange("add_function"), routing_key="add_function"),
    Queue("default", Exchange("default"), routing_key="default"),
)

#SQS에서는 필요없음
app.conf.task_routes = {
    "myapp.tasks.add_test": {"queue": "add_function"},
}

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

import time
from celery_app import app
from django.utils import timezone
from .models import TaskResult

@app.task(bind=True)
def add_test(self, x, y):
    start_time = time.time()

    # 작업 로직
    result = x + y
    task_id = self.request.id

	# 사실상 이것 때문에 result backend가 대체된것.
    TaskResult.objects.create(
        task_id=task_id,
        status="SUCCESS",
        result=result,
        date_done=timezone.now(),
    )
    return result

celery -A {작업 정의 모듈} worker -l info -Q {큐이름}

celery -A task worker -l info -Q add_function

from myapp.tasks import add_test

def run_celery():
    # 작업을 비동기로 호출
    result = add_test.apply_async((4, 6), queue="add_function")

    # 작업 ID 확인
    task_id = result.id
    print(f"Task ID: {task_id}")

    # 작업의 결과를 기다림
    try:
        result_value = result.wait(timeout=30)  # 최대 10초 동안 대기
        print(f"Task Result: {result_value}")
    except TimeoutError:
        print("The task is still running after 30 seconds.")

# 실행
run_celery()

pythonCopy code
@app.task(bind=True, autoretry_for=(Exception,), retry_kwargs={'max_retries': 5, 'countdown': 10})
def add_test(self, x, y):
    try:
        result = x + y
        task_id = self.request.id

        # 작업 결과를 데이터베이스에 저장
        TaskResult.objects.create(
            task_id=task_id,
            status="SUCCESS",
            result=result,
            date_done=timezone.now(),
        )

        return result
    except Exception as exc:
        raise self.retry(exc=exc)