Celery로 안정적인 비동기 서비스 만들기

Celery 운영 시 고려해야 할 노하우를 정리합니다.

1. Sentry로 작업 실패 감지하기

왜 필요한가?

• Celery는 기본적으로 오류를 콘솔에만 출력합니다.
• 서비스가 커지면 로그만으로 오류를 추적하기 어렵습니다.

연동 방법

1. Sentry SDK 설치

pip install sentry-sdk

1. Django 설정에 추가

import sentry_sdk
from sentry_sdk.integrations.django import DjangoIntegration

sentry_sdk.init(
    dsn="https://<your-key>@sentry.io/<project-id>",
    integrations=[DjangoIntegration()],
    traces_sample_rate=1.0,
    send_default_pii=True
)

1. Celery Task 실패 시 Sentry로 전송하기

from celery.signals import task_failure

@task_failure.connect
def handle_task_failure(sender=None, exception=None, **kwargs):
    from sentry_sdk import capture_exception
    capture_exception(exception)

실시간 오류 추적과 Slack 연동까지 확장 가능

2. 작업 중복 방지: debounce와 Redis Lock

문제 상황

• 사용자가 버튼을 여러 번 누르거나, 같은 API가 중복 호출되면 동일 작업이 여러 번 실행됨
• 이메일 중복 전송, 중복 크롤링 등 서비스 품질 저하 발생

debounce 처리 예시 (시간 기준)

_last_sent = {}

@shared_task
def notify_user(user_id):
    now = time.time()
    if _last_sent.get(user_id, 0) + 30 > now:
        return  # 30초 내 중복 실행 방지
    _last_sent[user_id] = now
    send_email(user_id)

Redis 기반 Lock 예시 (보다 안전한 방식)

import redis
r = redis.Redis()

@shared_task(bind=True)
def safe_task(self, task_id):
    lock_key = f"lock:{task_id}"
    with r.lock(lock_key, timeout=30):
        do_something()

3. Celery 로그 커스터마이징

로그 포맷 설정 (settings.py)

CELERYD_HIJACK_ROOT_LOGGER = False
CELERYD_LOG_FORMAT = "[%(asctime)s:%(levelname)s] [%(task_name)s] %(message)s"
CELERYD_TASK_LOG_FORMAT = "[%(asctime)s:%(levelname)s] Task %(task_name)s[%(task_id)s]: %(message)s"

작업별 로그 출력 예시

@shared_task(bind=True)
def process(self):
    self.logger.info("작업 시작")

• Task 이름과 ID를 함께 출력하여 추적 가능성 향상

4. 운영 중 자주 발생하는 이슈 대응 전략

문제 상황	대응 방법
워커 중단	systemd/supervisor로 재시작 자동화
Redis 연결 끊김	retry/backoff, 재시도 횟수 제한
작업 무한 대기	task_time_limit, soft_time_limit 설정
작업 누락	결과 백엔드 활용, Flower 모니터링 병행