Flask Celery 비동기

✅ 기본 구성

1. Flask 앱 생성
2. Celery 구성
3. 작업(Task) 정의
4. 비동기 호출

---

🔧 1. 설치

pip install flask celery redis

Redis를 브로커로 사용 (RabbitMQ도 가능)

---

📦 2. 프로젝트 구조 예시

 

myapp/
├── app.py
├── celery_worker.py
└── tasks.py

---

🧩 3. 코드 예시

app.py – Flask 앱

from flask import Flask, request, jsonify
from tasks import long_task

app = Flask(__name__)

@app.route('/run-task', methods=['POST'])
def run_task():
    task = long_task.delay()
    return jsonify({'task_id': task.id}), 202

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

---

tasks.py – Celery 작업 정의

from celery import Celery
import time

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

@celery.task
def long_task():
    print("Starting long task...")
    time.sleep(10)  # Simulate long task
    return "Task completed!"

---

celery_worker.py – 워커 실행용 스크립트

from tasks import celery

# 콘솔에서 실행할 때: celery -A celery_worker.celery worker --loglevel=info

---

▶️ 실행 방법

1. Redis 서버 실행
2. Celery 워커 실행

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

1. Flask 앱 실행

python app.py

1. API 요청

curl -X POST http://localhost:5000/run-task

---

📌 참고사항

• .delay()는 비동기 호출입니다. .get()을 쓰면 동기적으로 결과를 기다립니다.
• 프로덕션에서는 Flask와 Celery를 분리 배포하고, Flower 등으로 모니터링할 수 있습니다.ㅍㅍㅍㅍㅍㅍㅍㅍㅍㅍㅍ