django-celery

Django + Celery 비동기 태스크 패턴

Redis 또는 RabbitMQ와 함께 Celery를 사용하여 Django에서 백그라운드 태스크를 처리하는 프로덕션 등급 패턴입니다.

활성화 시점

• Django 애플리케이션에 백그라운드 작업 또는 비동기 처리 추가
• 주기적/예약된 태스크 구현
• 요청 처리 사이클에서 느린 작업(이메일, PDF 생성, API 호출) 분리
• cron과 유사한 스케줄링을 위한 Celery Beat 설정
• 태스크 실패, 재시도 또는 큐 백로그 디버깅
• Celery 태스크 테스트 작성

프로젝트 설정

설치

pip install 'celery[redis]' django-celery-results django-celery-beat

celery.py — 앱 진입점

# config/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.development')

app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()  # INSTALLED_APP의 tasks.py 자동 발견

@app.task(bind=True, ignore_result=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

# config/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Django 설정

# config/settings/base.py

# 브로커 (프로덕션에서는 Redis 권장)
CELERY_BROKER_URL = env('CELERY_BROKER_URL', default='redis://localhost:6379/0')
CELERY_RESULT_BACKEND = env('CELERY_RESULT_BACKEND', default='django-db')

# 직렬화
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# 태스크 동작
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60        # 하드 제한: 30분
CELERY_TASK_SOFT_TIME_LIMIT = 25 * 60   # 소프트 제한: SoftTimeLimitExceeded 발생
CELERY_WORKER_PREFETCH_MULTIPLIER = 1   # 워커가 긴 태스크를 독점하지 않도록 방지
CELERY_TASK_ACKS_LATE = True            # 워커 충돌 시 재큐잉

# 결과 영속성
CELERY_RESULT_EXPIRES = 60 * 60 * 24   # 결과 24시간 유지

# Beat 스케줄러 (주기적 태스크용)
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

# 설치된 앱
INSTALLED_APPS += [
    'django_celery_results',
    'django_celery_beat',
]

워커 실행

# 워커 시작 (개발)
celery -A config worker --loglevel=info

# Beat 스케줄러 시작 (주기적 태스크)
celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler

# 워커 + Beat 결합 (개발 전용, 프로덕션 절대 금지)
celery -A config worker --beat --loglevel=info

# 프로덕션: 여러 워커와 동시성 설정
celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority

태스크 설계 패턴

기본 태스크

# apps/notifications/tasks.py
from celery import shared_task
import logging

logger = logging.getLogger(__name__)

@shared_task(name='notifications.send_welcome_email')
def send_welcome_email(user_id: int) -> None:
    """새로 등록된 사용자에게 환영 이메일을 보냅니다."""
    from apps.users.models import User
    from apps.notifications.services import EmailService

    try:
        user = User.objects.get(pk=user_id)
    except User.DoesNotExist:
        logger.warning('send_welcome_email: 사용자 %s를 찾을 수 없음', user_id)
        return  # 멱등성 유지 — 예외를 발생시키지 않고, 태스크는 이미 완료 불가능

    EmailService.send_welcome(user)
    logger.info('%s 사용자에게 환영 이메일 전송 완료', user_id)

재시도 가능한 태스크

@shared_task(
    bind=True,
    name='integrations.sync_to_crm',
    max_retries=5,
    default_retry_delay=60,       # 첫 재시도 전 대기 시간(초)
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,           # 지수 백오프
    retry_backoff_max=600,        # 최대 10분
    retry_jitter=True,            # 무작위 지연으로 군집 요청 방지
)
def sync_contact_to_crm(self, contact_id: int) -> dict:
    """일시적 오류 발생 시 재시도하며 외부 CRM에 연락처를 동기화합니다."""
    from apps.crm.services import CRMClient

    try:
        result = CRMClient().sync(contact_id)
        return result
    except CRMClient.RateLimitError as exc:
        # 응답 헤더의 특정 재시도 지연 시간 사용
        raise self.retry(exc=exc, countdown=int(exc.retry_after))

멱등성 태스크 패턴

동일한 입력으로 여러 번 실행해도 안전하도록 태스크 설계:

@shared_task(name='orders.mark_shipped')
def mark_order_shipped(order_id: int, tracking_number: str) -> None:
    """주문 배송 완료 표시 — 여러 번 실행해도 안전합니다."""
    from apps.orders.models import Order

    updated = Order.objects.filter(
        pk=order_id,
        status=Order.Status.PROCESSING,    # 가드: 이미 배송 완료된 주문은 업데이트하지 않음
    ).update(
        status=Order.Status.SHIPPED,
        tracking_number=tracking_number,
    )

    if not updated:
        logger.info('mark_order_shipped: 주문 %s가 이미 배송 완료되었거나 찾을 수 없음', order_id)

소프트 타임 리밋을 가진 태스크

from celery.exceptions import SoftTimeLimitExceeded

@shared_task(
    bind=True,
    name='reports.generate_pdf',
    soft_time_limit=120,
    time_limit=150,
)
def generate_pdf_report(self, report_id: int) -> str:
    """우아한 타임아웃 처리가 포함된 PDF 보고서 생성."""
    from apps.reports.services import PDFGenerator

    try:
        path = PDFGenerator.build(report_id)
        return path
    except SoftTimeLimitExceeded:
        # 하드 킬 전에 부분 파일 정리
        PDFGenerator.cleanup(report_id)
        raise

태스크 호출

from datetime import timedelta
from django.utils import timezone

# Fire and forget (비동기)
send_welcome_email.delay(user.pk)

# 미래에 예약
send_reminder.apply_async(args=[user.pk], countdown=3600)  # 1시간 후
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))

# 큐 라우팅과 함께 호출
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')

# 동기 실행 (테스트/디버깅 전용)
result = generate_pdf_report.apply(args=[report.pk])

Beat 스케줄링 (주기적 태스크)

코드 정의 스케줄

# config/settings/base.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'users.cleanup_expired_sessions',
        'schedule': crontab(hour=2, minute=0),   # 매일 오전 2시
    },
    'sync-inventory': {
        'task': 'products.sync_inventory',
        'schedule': 60.0,                         # 60초마다
    },
    'weekly-digest': {
        'task': 'notifications.send_weekly_digest',
        'schedule': crontab(day_of_week='monday', hour=8, minute=0),
    },
}

데이터베이스 정의 스케줄 (django-celery-beat 사용)

# Django 관리자 또는 코드에서 주기적 태스크 관리
from django_celery_beat.models import PeriodicTask, CrontabSchedule
import json

schedule, _ = CrontabSchedule.objects.get_or_create(
    hour='*/6', minute='0',
    timezone='UTC',
)

PeriodicTask.objects.update_or_create(
    name='6시간마다 재고 동기화',
    defaults={
        'crontab': schedule,
        'task': 'products.sync_inventory',
        'args': json.dumps([]),
        'enabled': True,
    }
)

Canvas: 태스크 체이닝 및 그룹화

from celery import chain, group, chord

# 체인: 결과를 전달하며 태스크를 순차적으로 실행
pipeline = chain(
    fetch_data.s(source_id),
    transform_data.s(),          # fetch_data의 결과를 첫 번째 인수로 받음
    load_to_warehouse.s(),
)
pipeline.delay()

# 그룹: 태스크를 병렬로 실행
parallel = group(
    send_welcome_email.s(user_id)
    for user_id in new_user_ids
)
parallel.delay()

# 코드: 병렬 태스크 + 모든 태스크 완료 시 콜백
result = chord(
    group(process_chunk.s(chunk) for chunk in data_chunks),
    aggregate_results.s(),       # 청크 결과 리스트와 함께 호출됨
)
result.delay()

오류 처리 및 데드 레터 큐

# apps/core/tasks.py
from celery.signals import task_failure

@task_failure.connect
def on_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
    """모든 태스크 실패를 Sentry / 알림에 기록합니다."""
    import sentry_sdk
    with sentry_sdk.new_scope() as scope:
        scope.set_context('celery', {
            'task': sender.name,
            'task_id': task_id,
            'args': args,
            'kwargs': kwargs,
        })
        sentry_sdk.capture_exception(exception)

# 최대 재시도 후 실패한 태스크를 데드 레터 큐로 라우팅
@shared_task(
    bind=True,
    max_retries=3,
    name='payments.charge_card',
)
def charge_card(self, order_id: int) -> None:
    from apps.payments.models import Order, FailedCharge

    try:
        _do_charge(order_id)
    except Exception as exc:
        if self.request.retries >= self.max_retries:
            # 수동 검토를 위해 데드 레터 테이블에 저장
            FailedCharge.objects.create(
                order_id=order_id,
                error=str(exc),
                task_id=self.request.id,
            )
            return  # 예외를 발생시키지 않음 — 태스크가 영구적으로 실패함
        raise self.retry(exc=exc)

Celery 태스크 테스팅

유닛 테스트 (브로커 없음)

# tests/test_tasks.py
import pytest
from unittest.mock import patch, MagicMock
from apps.notifications.tasks import send_welcome_email

class TestSendWelcomeEmail:

    @pytest.mark.django_db
    def test_sends_email_to_existing_user(self, user):
        with patch('apps.notifications.services.EmailService') as mock_email:
            send_welcome_email(user.pk)
            mock_email.send_welcome.assert_called_once_with(user)

    @pytest.mark.django_db
    def test_skips_missing_user_gracefully(self):
        """큐에 넣은 후 실행 사이에 사용자가 삭제되어도 예외가 발생하지 않아야 합니다."""
        send_welcome_email(99999)  # 존재하지 않는 사용자 — 예외가 발생하지 않아야 함

CELERY_TASK_ALWAYS_EAGER를 사용한 통합 테스트

# config/settings/test.py
CELERY_TASK_ALWAYS_EAGER = True      # 테스트에서 태스크를 동기적으로 실행
CELERY_TASK_EAGER_PROPAGATES = True  # 태스크 예외를 다시 발생시킴

# tests/test_integration.py
@pytest.mark.django_db
def test_registration_triggers_welcome_email(client):
    with patch('apps.notifications.services.EmailService') as mock_email:
        response = client.post('/api/users/', {
            'email': '[email protected]',
            'password': 'strongpass123',
        })

    assert response.status_code == 201
    mock_email.send_welcome.assert_called_once()

재시도 테스팅

@pytest.mark.django_db
def test_task_retries_on_connection_error():
    with patch('apps.crm.services.CRMClient.sync') as mock_sync:
        mock_sync.side_effect = ConnectionError('timeout')

        with pytest.raises(ConnectionError):
            sync_contact_to_crm.apply(args=[1], throw=True)

        assert mock_sync.call_count == 1  # eager 모드에서는 첫 번째 시도만 실행됨

모니터링

# 활성 워커 및 큐 확인
celery -A config inspect active
celery -A config inspect stats
celery -A config inspect reserved

# 큐 길이 확인 (Redis)
redis-cli llen celery

# Flower: 웹 기반 실시간 모니터러
pip install flower
celery -A config flower --port=5555

안티 패턴

# BAD: 모델 인스턴스 전달 — 실행 시점에 오래된 데이터일 수 있음
send_welcome_email.delay(user)        # ORM 객체를 절대 전달하지 마세요
send_welcome_email.delay(user.pk)     # 항상 PK를 전달하세요

# BAD: 프로덕션 뷰에서 태스크를 동기적으로 호출
result = generate_report.apply()      # 요청 스레드를 차단합니다

# BAD: 가드 없는 비멱등성 태스크
@shared_task
def charge_and_fulfill(order_id):
    order.charge()     # 태스크가 재시도되면 두 번 청구될 수 있습니다!
    order.fulfill()

# GOOD: 상태 가드가 있는 멱등성 태스크
@shared_task
def charge_and_fulfill(order_id):
    order = Order.objects.select_for_update().get(pk=order_id)
    if order.status != Order.Status.PENDING:
        return  # 이미 처리됨
    order.charge()
    order.fulfill()

프로덕션 체크리스트

확인 사항	설정
워커 충돌 시 재시작	supervisord 또는 systemd 유닛
CELERY_TASK_ACKS_LATE = True	워커 충돌 시 태스크 재큐잉
CELERY_WORKER_PREFETCH_MULTIPLIER = 1	긴 태스크의 공정한 분배
우선순위 별 별도 큐	-Q default,high_priority,low_priority
CELERY_TASK_SOFT_TIME_LIMIT 설정	하드 킬 전 우아한 타임아웃
Sentry 통합	모든 task_failure 신호 캡처
Flower 또는 기타 모니터	큐 깊이 가시성 확보
Beat는 단일 노드에서만 실행	중복 예약 태스크 실행 방지

관련 스킬

• django-patterns — ORM, 서비스 레이어 및 프로젝트 구조
• django-tdd — Django 모델, 뷰 및 서비스 테스팅
• python-testing — pytest 설정 및 픽스처