Celery 태스크 라우팅과 신뢰성

큐를 어떻게 나누느냐가 처리량을 정하고, 확인 응답을 언제 보내느냐가 유실과 중복을 가릅니다.

한 큐에 모든 일을 몰아넣을 때

Celery는 파이썬에서 널리 쓰는 분산 태스크 큐로, 프로듀서가 메시지를 브로커에 넣으면 워커가 꺼내 실행합니다. 기본 설정은 모든 태스크를 하나의 큐에 넣습니다. 이메일 발송과 영상 인코딩이 같은 줄에 서면, 수십 초짜리 인코딩이 짧은 메일 작업을 막습니다.

해결의 출발점은 작업 성격에 따라 큐를 나누는 것입니다. 짧고 잦은 작업과 길고 무거운 작업을 다른 큐에 두고, 큐마다 워커를 따로 붙입니다. 그러면 한쪽이 밀려도 다른 쪽 지연이 번지지 않습니다.

Exchange가 메시지를 큐로 보내는 방식

큐 분리는 라우팅 규칙으로 구현합니다. Celery는 고급 메시지 큐 프로토콜(Advanced Message Queuing Protocol, AMQP) 모델을 따릅니다. 메시지는 큐로 바로 가지 않고 Exchange를 거칩니다. Exchange는 라우팅 키(routing key)와 바인딩 규칙을 보고 어느 큐에 넣을지 정합니다.

Exchange는 라우팅 방식에 따라 네 종류로 나뉩니다. 어떤 종류를 쓰느냐가 큐를 고르는 규칙을 정합니다.

Exchange 종류	라우팅 방식	쓰임
Direct	routing_key 정확 일치	큐를 1:1로 지정
Topic	와일드카드 패턴 (`feed.*`, `feed.#`)	카테고리별 분기
Fanout	바인딩된 모든 큐에 복제	브로드캐스트
Headers	메시지 헤더 속성 일치	키 외 조건으로 분기

Topic Exchange의 feed.#는 feed.로 시작하는 모든 키를 받는 바인딩 패턴입니다. 발행할 때는 feed.sync처럼 구체적인 키를 씁니다. 큐 정의와 라우팅 규칙은 다음처럼 설정합니다.

python

from celery import Celery
from kombu import Exchange, Queue
 
app = Celery('myapp')
 
default_exchange = Exchange('tasks', type='direct')
feed_exchange = Exchange('feeds', type='topic')
 
app.conf.task_queues = (
    Queue('default', exchange=default_exchange, routing_key='celery'),
    Queue('priority', exchange=default_exchange, routing_key='priority',
          queue_arguments={'x-max-priority': 10}),
    Queue('feed_tasks', exchange=feed_exchange, routing_key='feed.#',
          durable=True),
)
 
app.conf.task_routes = {
    'myapp.tasks.high_priority': {'queue': 'priority', 'routing_key': 'priority'},
    'feeds.*': {'queue': 'feed_tasks', 'routing_key': 'feed.sync'},
}

task_routes는 태스크 이름 패턴을 큐에 매핑하는 방식으로, 와일드카드와 함수형 라우터를 모두 지원합니다. 코드에서 apply_async(queue=...)로 큐를 직접 지정할 수도 있지만, 규칙을 설정에 모아 두면 호출부가 큐 이름을 몰라도 됩니다.

확인 응답 시점이 가르는 유실과 중복

라우팅이 처리량 문제라면, 확인 응답(acknowledgement, 이하 ack)은 신뢰성 문제입니다. 워커는 메시지를 처리했다는 ack를 브로커에 보내고, 브로커는 ack를 받아야 그 메시지를 큐에서 지웁니다. ack를 언제 보내느냐가 두 갈래로 갈립니다.

항목	Early Ack (기본)	Late Ack
ack 시점	메시지 수신 직후	태스크 실행 완료 후
워커 크래시	태스크 유실	재전송, 중복 실행 가능
속도	빠름	ack가 지연됨
전제 조건	유실 용인	멱등성

기본값인 Early Ack은 메시지를 받자마자 ack를 보냅니다. 워커가 실행 도중 죽으면 그 태스크는 사라집니다. Late Ack은 실행을 끝낸 뒤 ack를 보내므로, 크래시가 나면 브로커가 같은 메시지를 다른 워커에 다시 보냅니다.

python

app.conf.task_acks_late = True
app.conf.task_reject_on_worker_lost = True
app.conf.worker_prefetch_multiplier = 1

task_acks_late=True로 유실을 막고, task_reject_on_worker_lost=True로 워커가 비정상 종료해도 메시지를 다시 큐에 넣게 합니다. 그 대가로 중복 실행을 감수해야 합니다.

중복 실행을 흡수하는 멱등성

Late Ack를 켜는 순간 전달 보장은 최소 한 번(at-least-once)이 됩니다. 같은 메시지가 두 번 실행될 수 있다는 뜻입니다. 중복은 분산 큐에서 정상적으로 일어나는 일이며, 원인은 한 가지가 아닙니다.

상황	원인	결과
워커 크래시	실행 중 ack 전에 죽음	재전송 후 중복 실행
네트워크 단절	완료했으나 ack 미전송	재전송 후 중복 실행
Visibility timeout 만료	처리가 timeout보다 김 (Redis)	재발행 후 중복 실행
브로커 페일오버	복제 전 노드 죽음	다른 노드가 재발행

Redis 브로커는 RabbitMQ와 달리 AMQP의 확인 응답을 그대로 갖지 않아, visibility timeout으로 흉내냅니다. 워커가 메시지를 가져가도 timeout(기본 3,600초) 안에 ack가 없으면 브로커가 다시 발행합니다. 그래서 처리 시간이 timeout을 넘는 긴 작업은 중복 위험이 큽니다.

python

app.conf.broker_transport_options = {
    'visibility_timeout': 3600,  # Redis 전용, 초 단위
}

중복을 없앨 수 없으니, 같은 작업을 여러 번 실행해도 결과가 같도록 멱등성(idempotency)을 설계합니다. 가장 견고한 방법은 데이터베이스 unique 제약입니다. task_id를 멱등성 키로 저장하고, 두 번째 실행이 IntegrityError를 만나면 기존 결과를 돌려줍니다.

python

from django.db import IntegrityError, transaction
 
@app.task(bind=True, acks_late=True)
def process_payment(self, order_id):
    try:
        with transaction.atomic():
            payment = Payment.objects.create(
                order_id=order_id,
                idempotency_key=self.request.id,  # task_id를 키로 사용
                amount=calculate_amount(order_id),
            )
            charge_credit_card(payment)
            return {'status': 'success', 'payment_id': payment.id}
    except IntegrityError:
        existing = Payment.objects.get(idempotency_key=self.request.id)
        return {'status': 'duplicate', 'payment_id': existing.id}

재시도도 같은 전제 위에 섭니다. autoretry_for로 예외를 잡고, retry_backoff=True로 재시도 간격을 1초, 2초, 4초처럼 지수로 늘립니다. 멱등하지 않은 태스크에 재시도를 붙이면 부분 실행이 쌓여 중복이 더 늘어납니다.

python

@app.task(bind=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def fetch_url(self, url: str) -> str:
    import requests
    r = requests.get(url, timeout=10)
    r.raise_for_status()
    return r.text[:2000]

prefetch와 직렬화

워커는 한 번에 여러 메시지를 미리 받아 둡니다. worker_prefetch_multiplier가 그 개수를 정하며, 기본값 4는 짧은 작업에 맞춰져 있습니다. 긴 작업에서는 1로 낮춰, 한 워커가 미리 잡아 둔 작업이 다른 워커를 굶기지 않게 합니다.

직렬화는 보안과 직결됩니다. 기본값 JSON은 데이터만 옮기므로 안전합니다. Pickle은 파이썬 객체를 그대로 실어 임의 코드 실행 위험이 있으므로, accept_content=['json']으로 막고 워커를 root 권한으로 띄우지 않습니다. prefetch와 직렬화는 각각 처리량과 안전성에 관한 설정이지만, 라우팅·ack·멱등성과 함께 한 묶음으로 정합니다.

정리

큐 분리는 처리량을, 확인 응답 시점은 신뢰성을 결정합니다. Late Ack로 유실을 막으면 그 대가로 중복 실행을 받아들여야 하고, 멱등성과 지수 백오프 재시도가 그 중복을 흡수합니다. Redis를 쓸 때는 visibility timeout이 긴 작업에서 또 다른 중복 원인이 되므로 작업 시간과 함께 조정합니다. 라우팅, ack, 멱등성, prefetch, 직렬화는 따로 노는 설정이 아니라 한 신뢰성 묶음으로 함께 정해야 합니다.