Celery 태스크 라우팅과 신뢰성

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

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

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

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

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

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

flowchart LR
    P[프로듀서<br/>apply_async 호출] --> E{Exchange<br/>라우팅 규칙 적용}
    E -->|routing_key=priority| Q1[priority 큐<br/>빠른 워커]
    E -->|routing_key=feed.sync| Q2[feed_tasks 큐<br/>피드 워커]
    E -->|기본| Q3[default 큐<br/>일반 워커]

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

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

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은 메시지를 받자마자 ack를 보냅니다. 워커가 실행 도중 죽으면 그 태스크는 사라집니다. Late Ack은 실행을 끝낸 뒤 ack를 보내므로, 크래시가 나면 브로커가 같은 메시지를 다른 워커에 다시 보냅니다.

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로 워커가 비정상 종료해도 메시지를 다시 큐에 넣게 합니다. 그 대가로 중복 실행을 감수해야 합니다.

flowchart TD
    A[메시지 수신] --> B[태스크 실행]
    B --> C{성공 여부}
    C -->|성공| D[확인 응답<br/>acknowledge 전송]
    C -->|크래시| E[재전송<br/>다른 워커가 재실행]
    D --> F[결과 백엔드 저장]

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

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

Redis 브로커는 RabbitMQ의 AMQP 전달 상태 대신 visibility timeout으로 미확인 메시지를 관리합니다. 워커가 메시지를 가져간 뒤 기본 3,600초 안에 ack하지 않으면 다시 전달합니다(Celery Redis 문서). 처리 시간이 timeout을 넘는 작업은 중복 실행될 수 있습니다.

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

중복 전달을 전제로, 같은 작업을 여러 번 실행해도 외부 효과가 한 번만 남도록 멱등성(idempotency)을 설계합니다. 결제처럼 외부 시스템을 호출하는 작업은 데이터베이스 unique 제약만으로 부족합니다. 외부 결제사도 같은 멱등성 키에 같은 승인 결과를 돌려준다는 계약을 제공해야 합니다.

function process_payment(order_id, idempotency_key):
    payment = database.get_or_create_payment(
        unique_key=idempotency_key,
        order_id=order_id,
        status="pending"
    )
 
    charge = payment_provider.charge(
        amount=payment.amount,
        idempotency_key=idempotency_key
    )
 
    database.mark_succeeded(
        unique_key=idempotency_key,
        provider_charge_id=charge.id
    )
    return database.get_payment(unique_key=idempotency_key)

외부 승인 뒤 mark_succeeded 전에 워커가 죽으면 로컬 상태는 pending에 머뭅니다. 재전달된 작업은 결제사에 같은 키를 보내므로 새 승인을 만들지 않고 기존 승인 결과를 받습니다. 이 계약이 없는 결제사는 조회와 대사, 보상 처리를 별도로 갖춰야 합니다. Celery도 late ack 태스크가 여러 번 실행될 수 있으므로 멱등성을 권고합니다(Celery 태스크 문서).

재시도도 같은 전제 위에 섭니다. retry_backoff=True는 1초, 2초, 4초의 상한을 지수로 늘리며, 기본 jitter가 실제 지연을 그 상한 안에서 무작위로 정합니다(Celery 태스크 문서). 멱등하지 않은 태스크의 재시도는 부분 실행을 누적할 수 있습니다.

@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, 직렬화는 서로 영향을 주므로 신뢰성 요구에 맞춰 함께 정해야 합니다.