Celery와 Redis를 이용한 스케일링

Source: Dev.to

동기식 인덱싱의 문제점

사용자가 YouTube 재생목록을 인덱싱하기 시작했을 때, 백엔드는 동기식 루프를 실행하여 각 비디오를 차례대로 처리했습니다.
전사본을 가져오고 인덱싱하는 데 약 1.5 초가 걸리므로, 1,000개의 비디오가 있는 재생목록은 약 25 분이 필요합니다. 브라우저는 약 60 초 후에 타임아웃되고, UI가 멈추며, 앱은 결국 크래시됩니다.

왜 작업 큐와 백그라운드 워커가 필요한가?

완전한 동기식 설계는 메인 요청‑응답 사이클을 차단합니다:

• 장시간 실행되는 작업이 웹 서버 스레드를 잡아둡니다.
• 요청이 쌓이고, 지연 시간이 증가하며, 시스템이 연결을 끊거나 크래시될 수 있습니다.

작업 큐는 작업을 요청 경로 밖으로 이동시킵니다:

1. HTTP 핸들러가 의도를 기록하고 작업을 큐에 넣습니다.
2. 즉시 클라이언트에 응답을 반환합니다.
3. 별도의 워커 프로세스가 큐에서 작업을 소비하고 병렬(또는 동시)로 실행합니다.

이점

• 웹 계층이 응답성을 유지합니다.
• 워커를 추가하기만 하면 확장이 간단합니다.
• 워커에서 발생한 실패가 전체 시스템에 전파되지 않습니다.
• 큐 수준에서 레이트‑리밋을 적용할 수 있어 YouTube IP 차단을 방지하는 데 유용합니다.

핵심 개념

Task

작업의 단일 단위. 이 프로젝트에서 작업은 process_video_task이며, 다음을 수행합니다:

• 비디오 ID를 받습니다.
• 프록시를 통해 전사본을 가져옵니다.
• 데이터를 Elasticsearch에 인덱싱합니다.

Producer

작업을 생성하는 애플리케이션 부분. 여기서는 Flask 엔드포인트가 재생목록 정보를 수집하고 작업을 Celery에 전송하여 즉시 작업을 오프로드합니다.

Broker

작업을 저장하고 생산자와 소비자 사이를 중재하는 소프트웨어.
Redis는 메시지 브로커이자 작업 상태를 추적하는 저장소로 사용되어 실시간 진행 상황 업데이트를 가능하게 합니다.

Consumer (Worker)

브로커에서 작업을 끌어와 실행하는 프로세스. Celery가 워커 프로세스, 메모리, 확인(acknowledgment), 재시도를 관리합니다.

구현 세부 사항

Celery Task (Python)

# tasks.py
from celery import Celery

app = Celery('youtube_indexer', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def process_video_task(self, video_data, index_name):
    try:
        transcript = get_video_transcript(video_data['id'])
        if index_video(index_name, video_data, transcript):
            return (video_data['id'], True)
    except Exception as e:
        # Celery will handle retries based on max_retries
        raise self.retry(exc=e, countdown=60)

Redis에 Task ID 저장하기

# utils.py
TASK_KEY_PREFIX = "playlist_task:"
task_id_key = f"{TASK_KEY_PREFIX}{playlist_id}"
redis_conn.set(task_id_key, task.id, ex=7200)  # expires in 2 hours

Producer Endpoint (Flask)

# routes.py
from flask import Blueprint, request, jsonify
from .tasks import process_video_task

bp = Blueprint('index', __name__)

@bp.post('/index_playlist')
def index_playlist():
    data = request.json
    playlist_id = data['playlist_id']
    videos = get_videos_from_playlist(playlist_id)

    for video in videos:
        process_video_task.delay(video, index_name='my_index')

    return jsonify({"status": "queued", "playlist_id": playlist_id})

다른 언어 생태계

생산자, 브로커, 소비자 역할은 생태계가 달라져도 동일합니다.

아키텍처 패턴: Fan‑Out (Orchestrator)

시스템은 Fan‑Out / Orchestrator 패턴을 따릅니다:

1. Manager(Flask 엔드포인트)가 index_playlist_task와 같은 단일 오케스트레이션 작업을 생성합니다.
2. 이 작업이 다수의 process_video_task 작업을 디스패치합니다—비디오당 하나씩.
3. 워커들이 fan out하여 비디오를 동시에 처리합니다.

이 분리 덕분에 초기 요청은 가볍게 유지되면서 무거운 작업은 병렬 처리됩니다.

정리

• 장시간 실행되는 작업을 작업 큐에 오프로드하면 요청 타임아웃을 방지하고 UI 응답성을 유지할 수 있습니다.
• Redis는 빠른 인‑메모리 브로커를 제공하고, Celery는 재시도, 모니터링, 워커 관리를 위한 오케스트레이션 레이어를 추가합니다.
• 동일한 원칙이 여러 언어에 적용됩니다—큐 라이브러리를 교체하되 생산자‑브로커‑소비자 모델은 유지합니다.

Celery와 Redis를 중심으로 앱을 재구성함으로써 대규모 재생목록 인덱싱이 신뢰성 있게, 확장 가능하게, 그리고 사용자 친화적으로 변했습니다.