[Django + Celery + Redis] 비동기 태스크 구현하기

Redis 설치

공식 Redis 홈페이지의 메뉴얼에 따라 Redis를 다운로드하고 컴파일합니다. 컴파일이 정상적으로 완료되었는지 확인하고 싶다면 테스트를 진행해볼 수도 있습니다.

$ wget http://download.redis.io/redis-stable.tar.gz
$ tar xvzf redis-stable.tar.gz
$ cd redis-stable
$ make
$ make test # optional

컴파일이 성공하였으면 설치를 진행합니다.

이 때, 설치는 PATH 환경변수가 /usr/local/bin 이라는 가정하에 진행됩니다. 그렇지 않다면, Option 2와 같은 방식으로 직접 복사를 하시면 됩니다.

# Option 1
$ sudo make install

# Option 2
$ sudo cp src/redis-server /usr/local/bin/
$ sudo cp src/redis-cli /usr/local/bin/

설치가 완료되었으면 redis-server를 실행합니다.

# Starting Redis
$ redis-server

아래와 비슷한 로그가 출력된다면 정상적으로 실행된 것입니다.

[28550] 01 Aug 19:29:28 # Warning: no config file specified, using the default config. In order to specify a config file use 'redis-server /path/to/redis.conf'
[28550] 01 Aug 19:29:28 * Server started, Redis version 2.2.12
[28550] 01 Aug 19:29:28 * The server is now ready to accept connections on port 6379
... more logs ...

정상적으로 서버가 실행되고 있는 지 확인하고 싶다면, redis-cli를 통해 아래와 같이 확인해볼 수 있습니다.

# Check if Redis is working
$ redis-cli ping
PONG

참고 : Redis - Quick Start

이제 Redis 서버는 준비가 완료되었습니다. Python 라이브러리인 Celery를 설치하고, Celery에 redis 설정값을 추가하면 Django에서도 비동기 태스크를 이용할 수 있습니다.

Django + Celery 설치 및 설정

먼저 Django와 Celery를 설치해봅시다. celery[redis]를 통해서 redis를 이용하기 위해 필요한 패키지를 포함하여 설치를 진행합니다.

# Install Django
$ pip install Django
...
Successfully installed Django-2.2.1 pytz-2019.1 sqlparse-0.3.0

# Install Django
$ pip install 'celery[redis]==4.3.0'
...
Successfully installed amqp-2.4.2 billiard-3.6.0.0 celery-4.3.0 kombu-4.5.0 redis-3.2.1 vine-1.3.0

Django 설치가 완료되었으면, 프로젝트를 하나 생성해줍시다.

$ django-admin startproject server

프로젝트를 생성하면, 아래와 같은 폴더 구조로 생성됩니다.

여기서 추가해야할 것은 celery를 실행할 celery.py입니다.

celery에 대한 설정값을 담기 위해서 settings.py를 수정해야하고, celery를 항상 import 하기 위해서 _init_.py에 코드를 추가하여야 합니다.

server/
    manage.py
    server/
      M __init__.py  
      M settings.py 
        urls.py
        wsgi.py
      + celery.py

먼저, server 폴더 내에 celery.py를 생성합니다.

# server/celery.py
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

# https://stackoverflow.com/questions/50336688/django-load-production-settings-for-celery
# set the default Django settings module for the 'celery' program.
# os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')

# server를 기준으로 Celery 실행
app = Celery('server')

# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()


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

그 다음, _init_.py를 수정해서 디렉토리 내의 모든 python 파일들이 celery_app을 import 하도록 합니다.

# server/__init__.py
from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

다음은 settings.py 입니다. 필수적으로 BROKER_URL을 선언하여 redis 서버와 연결하고, CELERY_RESULT_BACKEND를 통해서 결과값을 저장해둘 백엔드 서버도 연결합니다. 이외에도 여러 설정 값을 이 부분에서 세팅합니다. (설정가이드 참조)

# server/settings.py

# Required
BROKER_URL = 'redis://localhost:6379/0' # your redis server url
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' # your redis url for getting result

# Customize the settings
# https://docs.celeryproject.org/en/latest/userguide/configuration.html
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TAST_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Seoul' # Define the timezone for the scheduler, Celery beat.

비동기 태스크

모든 기본 설정은 완료되었습니다. 이제 앱을 추가하고, 앱 디렉토리에 tasks.py를 추가하여 태스크들을 정의하여 사용하면 됩니다.

먼저 Django 앱을 추가하여 줍니다.

python manage.py startapp app

이제 이런 디렉토리 구조가 되었습니다. 여기서 app 디렉토리에 tasks.py를 추가합니다.

server/
    manage.py
    app/
        __init__.py
        admin.py
        apps.py
        models.py
      + tasks.py
        tests.py
        views.py
        migrations/
    server/
        __init__.py  
        settings.py 
        urls.py
        wsgi.py
        celery.py

tasks.py엔 다음과 같이 예제 코드를 작성합니다.

# server/app/tasks.py
# Create your tasks here
from __future__ import absolute_import, unicode_literals
from celery import shared_task

@shared_task
def add(x, y):
    return x + y


@shared_task
def mul(x, y):
    return x * y


@shared_task
def xsum(numbers):
    return sum(numbers)

이제 완료되었으면, 아래 코드를 실행 시켜 celery를 실행하고, django가 실행된 shell을 실행합니다.

# set DJANGO_SETTINGS_MODULE
export DJANGO_SETTINGS_MODULE=server.settings
celery -A server worker -l info & > celery_log
python manage.py shell

위에서 추가했던 tasks.py를 import해서 add를 실행해보고, 정상적으로 동작하는지 확인합니다.

이 때, add(4,4)가 아닌 add.delay(4,4)로 작성합니다. 이에 대한 자세한 설명은 공식 문서를 참조해주세요.

Python 3.5.2 (default, Nov 12 2018, 13:43:14) 
[GCC 5.4.0 20160609] on linux
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from app.tasks import *
>>> result = add.delay(4, 4)
>>> result.get()
8

위와 같이 동작하였다면, celery_log내의 로그에서 아래와 비슷한 로그가 작성되었습니다. add 함수가 비동기적으로 실행되어 성공하였다는 내용입니다.

[2019-05-14 14:31:56,600: INFO/ForkPoolWorker-3] Task app.tasks.add[8cd7ebdb-9f0c-42d9-91c1-608a06caa692] succeeded in 0.0005141769997862866s: 8