문서가격블로그
문의하기
로그인무료로 가입하기
Back to Tech Articles

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

Jul 26, 2025
# Misc

James Reed

Infrastructure Engineer · Leapcell

이 게시물은 English, 日本語 및 Deutsch에서도 볼 수 있습니다.
Cover of "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>

구성 요소 실행

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

  1. Redis 시작 (아직 실행 중이 아닌 경우):

    redis-server

  2. Celery Worker 시작: 프로젝트 루트 디렉토리에서 새 터미널을 엽니다.

    celery -A myproject worker -l info

    워커가 Redis에 연결되어 작업을 처리할 준비가 되었다는 것을 나타내는 출력이 표시되어야 합니다.

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

    python manage.py runserver

  4. 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는 작업 실행에 대한 필수적인 실시간 가시성과 제어를 제공합니다. 이 강력한 삼각 편대는 애플리케이션 성능, 사용자 경험 및 운영 효율성을 크게 향상시켜 현대 웹 개발에 필수적인 패턴이 됩니다. 동기 처리와 장기 실행 작업을 분리함으로써 분산 아키텍처의 잠재력을 최대한 발휘하게 됩니다.

Share this article

HackerNews

More Posts from Leapcell

cover

Go Gin 애플리케이션에서 Viper를 사용한 설정 간소화

cover

Gin과 FastAPI에서의 미들웨어 실행 과정 심층 분석

Popular Posts