[Django & Redis & Celery] - 비동기 작업 큐

현재 진행하고 있는 프로젝트에서 Django를 사용하고있는데, 여러 창에서 한번에 작업을 요청하면 병목현상으로인해 터지는 문제가 발생하였습니다. 

이런 문제가 발생한 이유는 Django가 동기적인 작동을 하기 때문에, 요청된 작업이 끝날 때까지 다른일을 하지 않는 블로킹 문제 때문이었습니다.

 

원래 Google Cloud Run 환경을 사용하고 있었는데, 이 환경에서 비동기 처리를 구현하려면 Cloud Tasks, Pub/Sub, 또는 별도의 워커 서비스 등을 추가로 설정해야 했습니다. 이러한 복잡한 인프라 설정 대신, 비교적 간단하게 파이썬 내에서 통합 관리할 수 있는 Celery를 선택하게 되었습니다!

---

 

먼저, Celery란?

• 분산형 작업 큐 시스템으로, 주로 Python으로 작성된 애플리케이션의 장시간 실행 작업을 웹 요청-응답 주기로부터 분리하여 비동기적으로 처리하도록 도와줍니다.

 

Celery의 주요 역할

• 비동기 처리 : Django 웹 서버가 파일 업로드와 같은 HTTP 요청을 받으면 Celery에게 작업을 위임 + 웹 서버는 즉시 사용자에게 응답함. => 사용자 대기시간 최소화 + 웹 서버 부하 감소효과!
• 작업자 관리 : Celery Worker는 Redis에서 작업을 가져와 무거운 연산을 수행함.
• 스케줄링 : 정기적으로 실행되어야 하는 작업을 스케줄할수있음!

---

 

Redis란?

• 인메모리 데이터 구조 저장소로, DB, 캐시, 메시지 브로커로 사용된다.

 

Redis의 주요 역할

• 메시지 브로커 : Celery의 작업 요청이 잠시 대기하는 큐 역할. Celery Worker는 이 큐에서 작업을 꺼내 처리함!
• 백엔드 : Celery 작업의 결과와 상태를 저장하는 저장소 역할. 작업의 상태 저장
• 캐싱 : 웹 애플리케이션에서 자주 접근하는 데이터를 메모리에 저장하여 DB 부하를 줄이고 응답 속도를 높이는데 사용됨

 

Celery와 Redis의 연동

 

1. 요청 : web서비스의 views.py 함수가 apply_async를 호출

2. 대기 : 작업 요청은 Redis 브로커의 큐에 저장됨

3. 처리 : celery_worker는 Redis 큐를 감시하다가 새 작업이 들어오면 가져와 실행함

4. 상태 보고 : 작업 처리 중이나 완료시 상태 정보는 Redis 백엔드에 저장됨

5. 조회 : 클라이언트의 Polling요청을 받은 web서비스가 Redis 백엔드에서 작업상태를 조회하여 사용자에게 반환

 

---

그다음으로는 코드부분!

 

1. 프로젝트 환경 준비 (Docker Compose)

먼저 Celery 모듈을 설치해줍니다.

pip install Celery

 

그리고 Django, Celery, Redis를 분리된 컨테이너로 실행하기 위해 아래와 같은 코드를 추가합니다.

 

1-1) Docker Compose 설정(docker-compose.yml)

- 세가지 핵심 서비스 정의 + 파일 처리를 위해 Worker와 웹 서비스 컨테이너가 같은 물리적 저장 공간을 바라보도록 설정.

 

1-2) Redis 대기 (docker-entrypoint.sh)

- 앱/Worker 컨테이너가 시작되기 전, Redis가 완전히 준비될 때까지 기다림

#!/bin/bash

REDIS_HOST="redis_master" 

# Redis가 완전히 준비될 때까지 기다립니다.
echo "Waiting for Redis ($REDIS_HOST)..."
while ! redis-cli -h $REDIS_HOST ping; do
  sleep 1
done
echo "Redis started. Executing Gunicorn..."

# Web 컨테이너의 원래 CMD(Gunicorn)를 실행
exec "$@"

 

 

2. Celery Task 정의

- 무거운 연산을 @shared_task로 감싸서 Celery Worker가 실행할 수 있는 작업으로 만듬

- 작업이 완료되면 COMPLETED 상태로 업데이트하고 결과 파일 경로를 반환

# tasks.py
from celery import shared_task
....

@shared_task(bind=True, name="urls.py에서 정의한 Task 고유의 이름")
def exec_~~(self, job_id, in_path, original_filename):
    exec_update_job_status(job_id, 'PROCESSING')
    # ... 
    # ...
    exec_update_job_status(job_id, 'COMPLETED', result_path=pdf_path)
    return { "path" : pdf_path, "filename": download_name}

 

3. 작업 위임 및 API 엔드포인트 설정

- Django는 요청을 받으면 무거운 작업을 즉시 Celery Worker에게 넘김

 

1-1) 작업 위임

# views.py 
from .tasks import exec_ppt_to_pdf_task

def ppt_to_pdf(request):
    # ...

    try:
        # Celery Task 위임: 무거운 연산은 Worker에게 맡기고 바로 반환
        task_result = exec_~~~.apply_async(args=[job_id, in_path, f.name], task_id=job_id)
        
        # 즉시 응답
        return JsonResponse({
            "status": "Job accepted and processing",
            "job_id": job_id,
        }, status=202) # 202 Accepted 코드는 비동기 작업 접수 표준 응답입니다.
    except Exception as e:
        # ...

 

4. 클라이언트 상태 확인

- UI에서는 작업 ID를 사용하여 서버에 결과를 주기적으로 문의함

// ppt.html, docx.html, mask_fast.html 등에 포함된 JavaScript
function pollStatus(jobId) {
    // 2초마다 상태를 확인하는 타이머 설정
    const intervalId = setInterval(async () => {
        try {
            const res = await fetch(`/api/status/${jobId}/`);
            const data = await res.json();

            if (data.status === 'Completed') {
                clearInterval(intervalId); // 완료 시 폴링 중지

                const downloadUrl = `/api/download/${jobId}/`;

                // 다운로드 버튼 표시
                statusDiv.innerHTML = `
                    <a href="${downloadUrl}" class="btn btn-success mt-2 w-100">
                        📂 파일 다운로드 하기
                    </a>
                `;
            } else if (data.status === 'Failed') {
                clearInterval(intervalId); // 실패 시 폴링 중지
                showError("실패: " + data.message);
            }
        } catch (e) {
            // 통신 에러 처리
        }
    }, 2000); // 2초 간격
}

 

이를 통해 여러 창을 띄워놓고 한번에 작업을 실행하여도 빠르게 처리되는 결과를 얻을 수 있었습니다!