Django, Celery, Flower를 이용한 분산 작업 처리

소개

현대 웹 애플리케이션 환경에서 응답성과 확장성은 무엇보다 중요합니다. 사용자 상호 작용은 종종 이미지 처리, 데이터 분석, 대량 이메일 발송 또는 보고서 생성과 같은 복잡한 작업을 트리거합니다. 이러한 작업을 기본 요청-응답 주기 내에서 동기적으로 수행하면 사용자 경험이 느려지고, 타임아웃이 발생하며, 애플리케이션이 불안정해질 수 있습니다. 여기서 백그라운드 작업이라는 개념이 매우 중요해집니다. 시간이 많이 걸리는 작업을 별도의 프로세스로 분담함으로써, 사용자에게 빠르게 응답을 반환하여 인지된 성능을 향상시키고 애플리케이션이 더 많은 동시 요청을 처리하도록 할 수 있습니다. 이 글에서는 Django, Celery, Flower의 강력한 조합을 사용하여 이를 달성하는 방법과 분산 백그라운드 작업을 구축, 실행 및 모니터링하는 데 있어 이들의 시너지를 설명합니다.

핵심 개념 설명

구현 세부 정보에 들어가기 전에 관련 주요 기술에 대한 명확한 이해를 갖추도록 합시다.

Django: 신속한 개발과 깔끔하고 실용적인 디자인을 장려하는 고수준 Python 웹 프레임워크입니다. 우리 애플리케이션의 프런트엔드 역할을 하며 백그라운드 작업을 트리거하는 주체가 됩니다.
Celery: 분산 메시지 전달을 기반으로 하는 비동기 작업 큐/작업 큐입니다. Django 애플리케이션에서 별도의 워커 프로세스가 실행하도록 작업을 분담할 수 있게 해줍니다. 매우 유연하며 다양한 메시지 브로커를 지원하고 작업 스케줄링, 재시도, 속도 제한과 같은 기능을 제공합니다.
Broker: Django 애플리케이션(생산자)과 Celery 워커(소비자) 간의 통신을 용이하게 하는 메시지 큐입니다. 작업이 전송되면 브로커에 배치되고 워커가 거기서 가져와 처리합니다. 인기 있는 선택지로는 RabbitMQ와 Redis가 있습니다.
Celery Worker: 브로커에서 새 작업을 지속적으로 모니터링하는 별도의 프로세스입니다. 작업을 검색하면 해당 워커가 작업의 논리를 실행합니다. 동시 처리를 위해 여러 워커를 실행할 수 있습니다.
Celery Beat: Celery 큐에 주기적으로 작업을 디스패치하는 스케줄러입니다. 일일 데이터 백업이나 야간 보고서 생성과 같은 예약된 작업에 유용합니다.
Celery Flower: Celery를 위한 실시간 웹 기반 모니터입니다. 이 작업은 작업(대기 중, 시작됨, 성공, 실패), 워커 활동을 검사하는 사용자 친화적인 인터페이스를 제공하며 워커의 원격 제어도 가능하게 합니다.

분산 작업 구축, 실행 및 모니터링

실제 예제인 이미지 크기 조정을 통해 이러한 도구를 통합하는 과정을 살펴보겠습니다. 사용자가 이미지를 업로드하지만, 업로드 프로세스를 차단하는 대신 Celery 워커에 크기 조정을 분담하는 시나리오를 시뮬레이션합니다.

프로젝트 설정 및 설치

먼저 Django 프로젝트가 설정되어 있는지 확인합니다. 그런 다음 필요한 패키지를 설치합니다.

pip install Django celery redis flower Pillow

Redis는 단순성과 성능 때문에 Celery 브로커 및 결과 백엔드로 사용할 것입니다. Pillow는 이미지 조작을 위한 것입니다.

Django 프로젝트 구성

settings.py 파일을 수정하여 Celery를 구성합니다.

# myproject/settings.py

# Celery Configuration
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai' # 또는 현지 시간대

다음으로, Django 프로젝트의 루트 디렉토리(settings.py와 같은 레벨)에 celery.py 파일을 생성합니다.

# myproject/celery.py

import os
from celery import Celery

# 'celery' 프로그램에 대한 기본 Django 설정 모듈을 설정합니다.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

app = Celery('myproject')

# 여기에 문자열을 사용하면 워커가 설정 객체를 자식 프로세스로 직렬화할 필요가 없습니다.
# - namespace='CELERY'는 모든 celery 관련 설정 키가 `CELERY_` 접두사를 가져야 함을 의미합니다.
app.config_from_object('django.conf:settings', namespace='CELERY')

# 등록된 모든 Django 앱 구성에서 작업 모듈을 로드합니다.
app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

마지막으로, Django 앱이 Celery를 로드하도록 합니다. 프로젝트의 __init__.py 파일에서 이를 수행합니다.

# myproject/__init__.py

# Django가 시작될 때마다 항상 앱이 가져오도록 하여 shared_task가 이 앱을 사용하도록 합니다.
from .celery import app as celery_app

__all__ = ('celery_app',)

Celery 작업 정의

images라는 Django 앱을 생성하고 그 안에서 이미지 크기 조정 작업을 정의해 보겠습니다.

python manage.py startapp images

images 앱 내에 tasks.py 파일을 생성합니다.

# images/tasks.py

import os
from PIL import Image
from io import BytesIO
from django.core.files.base import ContentFile
from django.conf import settings
from .models import UploadedImage # 이미지 저장을 위한 모델이라고 가정
from celery import shared_task

@shared_task
def resize_image_task(image_id, max_width=800, max_height=600):
    try:
        uploaded_image = UploadedImage.objects.get(id=image_id)
        original_image_path = uploaded_image.image.path

        # 원본 이미지 열기
        img = Image.open(original_image_path)
        img.thumbnail((max_width, max_height), Image.Resampling.LANCZOS)

        # 크기가 조정된 이미지를 버퍼에 저장
        output_buffer = BytesIO()
        # 원본 형식 유지, 또는 JPEG와 같은 일반 형식 선택
        img_format = img.format if img.format else 'JPEG'
        quality = 85 if img_format == 'JPEG' else None
        img.save(output_buffer, format=img_format, quality=quality)
        output_buffer.seek(0)

        # 크기가 조정된 이미지에 대한 새 파일 이름 구성
        original_filename_base, original_filename_ext = os.path.splitext(uploaded_image.image.name)
        resized_filename = f"{original_filename_base}_resized{original_filename_ext}"

        # 모델을 조정된 이미지로 업데이트
        uploaded_image.resized_image.save(
            resized_filename,
            ContentFile(output_buffer.read()),
            save=False # 모델을 아직 저장하지 않음, 명시적으로 수행할 것임
        )
        uploaded_image.is_processed = True
        uploaded_image.save()
        
        return f"Image {image_id} resized successfully to {max_width}x{max_height}."
    except UploadedImage.DoesNotExist:
        return f"Error: Image with ID {image_id} not found."
    except Exception as e:
        return f"Error resizing image {image_id}: {str(e)}"

이미지를 저장하기 위한 간단한 Django 모델이 필요합니다.

# images/models.py
from django.db import models

class UploadedImage(models.Model):
    image = models.ImageField(upload_to='original_images/')
    resized_image = models.ImageField(upload_to='resized_images/', blank=True, null=True)
    uploaded_at = models.DateTimeField(auto_now_add=True)
    is_processed = models.BooleanField(default=False)

    def __str__(self):
        return f"Image {self.id}: {self.image.name}"

이제 마이그레이션을 실행해야 합니다. python manage.py makemigrations images 및 python manage.py migrate. 또한 이미지 저장을 위해 settings.py에서 MEDIA_ROOT 및 MEDIA_URL을 구성했는지 확인하십시오.

# myproject/settings.py
# ...
MEDIA_ROOT = os.path.join(BASE_DIR, 'media')
MEDIA_URL = '/media/'
# ...

Django에서 작업 트리거

이제 이미지 업로드를 처리하고 Celery 작업을 트리거하는 images 앱의 간단한 뷰를 만들어 보겠습니다.

# images/views.py
from django.shortcuts import render, redirect
from .forms import ImageUploadForm
from .models import UploadedImage
from .tasks import resize_image_task

def upload_image(request):
    if request.method == 'POST':
        form = ImageUploadForm(request.POST, request.FILES)
        if form.is_valid():
            uploaded_image = form.save()
            # 작업 큐에 넣기
            resize_image_task.delay(uploaded_image.id) 
            return redirect('image_list')
    else:
        form = ImageUploadForm()
    return render(request, 'images/upload.html', {'form': form})

def image_list(request):
    images = UploadedImage.objects.all().order_by('-uploaded_at')
    return render(request, 'images/list.html', {'images': images})

그리고 해당 폼:

# images/forms.py
from django import forms
from .models import UploadedImage

class ImageUploadForm(forms.ModelForm):
    class Meta:
        model = UploadedImage
        fields = ['image']

URL 설정:

# myproject/urls.py
from django.contrib import admin
from django.urls import path
from django.conf import settings
from django.conf.urls.static import static
from images.views import upload_image, image_list

urlpatterns = [
    path('admin/', admin.site.urls),
    path('upload/', upload_image, name='upload_image'),
    path('images/', image_list, name='image_list'),
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)

그리고 upload.html 및 list.html을 위한 몇 가지 기본 템플릿:

<!-- images/templates/images/upload.html -->
<h1>Upload Image</h1>
<form method="post" enctype="multipart/form-data">
    {% csrf_token %}
    {{ form.as_p }}
    <button type="submit">Upload</button>
</form>
<a href="{% url 'image_list' %}">View Images</a>

<!-- images/templates/images/list.html -->
<h1>Uploaded Images</h1>
<ul>
    {% for image in images %}
        <li>
            <p>Original: <img src="{{ image.image.url }}" width="100"></p>
            {% if image.is_processed %}
                <p>Resized: <img src="{{ image.resized_image.url }}" width="100"></p>
            {% else %}
                <p>Processing...</p>
            {% endif %}
            <p>Uploaded at: {{ image.uploaded_at }}</p>
        </li>
    {% empty %}
        <li>No images uploaded yet.</li>
    {% endfor %}
</ul>
<a href="{% url 'upload_image' %}">Upload New Image</a>

구성 요소 실행

이제 모든 것이 설정되었으므로 각 부분을 실행해 보겠습니다.

Redis 시작 (아직 실행 중이 아닌 경우):
```
redis-server
```
Celery Worker 시작: 프로젝트 루트 디렉토리에서 새 터미널을 엽니다.
```
celery -A myproject worker -l info
```
워커가 Redis에 연결되어 작업을 처리할 준비가 되었다는 것을 나타내는 출력이 표시되어야 합니다.

Django 개발 서버 시작: 다른 터미널을 엽니다.

python manage.py runserver

Flower 시작 (모니터링용): 세 번째 터미널을 엽니다.
```
celery -A myproject flower
```
Flower는 일반적으로 http://localhost:5555에서 액세스할 수 있습니다.

이제 브라우저에서 http://localhost:8000/upload/로 이동하여 이미지를 업로드하고 다음을 관찰하십시오.

Django 뷰는 작업이 분담되었음을 나타내는 빠른 리디렉션을 제공합니다.
Celery 워커 터미널은 resize_image_task를 수신하고 처리하는 로그를 보여줍니다.
Flower (http://localhost:5555)는 실시간으로 작업을 표시하고 작업 상태(PENDING, STARTED, SUCCESS)를 보여줍니다. 작업 ID를 클릭하여 자세한 정보, 인수 및 반환 값을 볼 수 있습니다.
처리 완료 후 http://localhost:8000/images/를 새로 고치면 원본 이미지와 크기 조정된 이미지가 모두 표시됩니다.

고급 개념 및 애플리케이션 시나리오

작업 체인/워크플로: Celery는 chord, group, chain 작업을 사용하여 복잡한 작업 워크플로를 지원합니다. 예를 들어, 이미지 크기를 조정한 후 클라우드 저장소에 업로드하는 작업, 그리고 데이터베이스 레코드를 업데이트하는 또 다른 작업을 체인으로 연결할 수 있습니다.
재시도: 작업은 사용자 정의 재시도 지연 및 최대 시도 횟수를 사용하여 실패 시 자동으로 재시도되도록 구성할 수 있어 애플리케이션이 더 복원력 있게 됩니다.
속도 제한: 워커가 특정 시간 프레임 내에서 실행할 수 있는 작업 수를 제어하여 타사 API 또는 중요 서비스에 대한 리소스 고갈을 방지합니다.

Celery Beat를 사용한 예약 작업: 특정 간격(예: 매일 오전 3시에 오래된 데이터를 정리하기 위해)으로 작업을 실행하려면 Celery Beat를 사용합니다.

# myproject/settings.py
CELERY_BEAT_SCHEDULE = {
    'clean-old-images-every-day': {
        'task': 'images.tasks.cleanup_old_images_task',
        'schedule': crontab(hour=3, minute=0), # 매일 오전 3시에 실행
    },
}

그런 다음 별도의 터미널에서 Celery Beat를 실행합니다: celery -A myproject beat -l info

오류 처리: try-except 블록을 사용하여 작업 내에서 적절한 오류 처리를 구현하고, 검사를 위해 오류를 내보내거나 로깅하는 것이 좋습니다.
동시성: 여러 작업 프로세스를 실행하거나 다른 시스템에 걸쳐 여러 작업 노드를 실행하여 작업자들의 규모를 조정하여 많은 양의 작업을 처리할 수 있습니다.

결론

Django와 Celery 및 Flower를 통합하면 백그라운드 작업을 관리하는 강력하고 확장 가능한 솔루션을 제공합니다. Celery는 시간이 많이 걸리는 작업을 분담하여 Django 애플리케이션이 응답성을 유지하도록 지원하고, Flower는 작업 실행에 대한 필수적인 실시간 가시성과 제어를 제공합니다. 이 강력한 삼각 편대는 애플리케이션 성능, 사용자 경험 및 운영 효율성을 크게 향상시켜 현대 웹 개발에 필수적인 패턴이 됩니다. 동기 처리와 장기 실행 작업을 분리함으로써 분산 아키텍처의 잠재력을 최대한 발휘하게 됩니다.