Flask에서 Celery를 사용한 비동기 작업 처리

Flask에서 Celery를 사용한 비동기 작업 처리

Flask는 가벼운 웹 프레임워크로, 비동기 작업 처리가 필요한 웹 애플리케이션에서 Celery와 함께 사용되면 더욱 강력한 기능을 제공합니다. 이 글에서는 Flask에서 Celery를 사용해 비동기 작업을 처리하는 방법을 단계별로 살펴보겠습니다.

---

1. 비동기 작업이란?

비동기 작업은 웹 애플리케이션에서 오래 걸리는 작업(예: 이메일 발송, 데이터 처리 등)을 별도의 프로세스에서 실행하여 사용자 경험을 향상시키는 기법입니다. 이를 통해 사용자는 작업 완료를 기다릴 필요 없이 빠르게 응답을 받을 수 있습니다.

2. Celery란 무엇인가?

Celery는 Python 기반의 분산 작업 큐로, 비동기 작업을 관리하는 데 사용됩니다. Celery는 작업 큐와 브로커를 사용하여 작업을 예약하고 실행합니다.

주요 구성 요소

• 브로커 (Broker): 작업 메시지를 큐에 저장합니다. (예: Redis, RabbitMQ)
• 워커 (Worker): 큐에서 작업을 가져와 실행합니다.
• 작업 (Task): 실행할 코드 조각입니다.

---

3. Flask와 Celery 통합하기

설치

Flask와 Celery를 설치합니다.

pip install flask celery redis

디렉토리 구조

다음과 같은 디렉토리 구조를 사용할 것입니다:

project/
│
├── app.py         # Flask 애플리케이션
├── celery_tasks.py # Celery 작업 정의
├── requirements.txt

---

Flask 애플리케이션 설정 (app.py)

from flask import Flask, jsonify
from celery import Celery

# Flask 애플리케이션 생성
def make_app():
    app = Flask(__name__)

    # Flask 설정
    app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
    app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

    # Celery 설정
    celery = make_celery(app)

    # 라우트 정의
    @app.route('/longtask', methods=['GET'])
    def longtask():
        task = long_running_task.apply_async()
        return jsonify({"task_id": task.id, "status": "Task started."}), 202

    return app

# Celery 생성
def make_celery(app):
    celery = Celery(
        app.import_name,
        broker=app.config['CELERY_BROKER_URL']
    )
    celery.conf.update(app.config)
    return celery

# Celery 작업 정의
@celery.task
def long_running_task():
    import time
    time.sleep(10)  # 10초 동안 대기
    return "Task completed!"

if __name__ == '__main__':
    app = make_app()
    app.run(debug=True)

Celery 작업 정의 (celery_tasks.py)

작업은 Celery 인스턴스를 통해 정의합니다.

from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@celery.task
def add(x, y):
    return x + y

---

4. Celery 워커 실행

Celery 워커를 실행하려면 터미널에서 다음 명령어를 입력합니다:

celery -A app.celery worker --loglevel=info

결과 확인

1. 웹 브라우저에서 http://localhost:5000/longtask를 호출합니다.
2. 터미널에서 Celery 워커의 로그를 확인합니다.

---

5. 결과 반환 처리

비동기 작업의 결과를 확인하려면 작업 ID를 사용합니다. 아래는 결과를 조회하는 API 예제입니다:

@app.route('/task_status/<task_id>', methods=['GET'])
def task_status(task_id):
    task = long_running_task.AsyncResult(task_id)
    return jsonify({
        "task_id": task.id,
        "status": task.status,
        "result": task.result
    })

---

6. 결론

Flask와 Celery를 사용하면 복잡한 비동기 작업을 손쉽게 처리할 수 있습니다. 이 글에서는 간단한 작업 큐를 설정하고 실행하는 과정을 살펴보았습니다. 실제 애플리케이션에서는 Celery를 사용해 이메일 발송, 데이터 처리, 대규모 계산 등 다양한 작업을 처리할 수 있습니다.

추가 참고 자료

• Flask 공식 문서
• Celery 공식 문서
• Redis 공식 문서