Django-Celery : First Steps with Celery

샐러리 튜토리얼 따라한 기록.

사용 환경 : Ubuntu 14.04 / python 3.5 / PyCharm 5.0.3 / Celery 3.1

pyenv 버전 만들기

>> pyenv virtualenv 3.5.0 celerytest

>> pyenv shell celerytest

>>pyenv version

version 확인

PyCharm에서 프로젝트 만들기

New Project :

Pure Python -> Interpreter : /home/ldy/.pyenv/versions/celerytest/bin/python3.5

pyenv가 설치된 폴더로 가서 interpreter을 선택한다.

Celery

http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html#first-steps

필요부분만 번역합니다. 오역, 의역 엄청 많음

1. 브로커 선택 - RabbitMQ

http://docs.celeryproject.org/en/latest/getting-started/brokers/rabbitmq.html#broker-rabbitmq

다운받기

>> sudo apt-get install rabbimq-server

서버 시작

>> sudo rabbitmq-server

백그라운드에서 시작

>> sudo rabbitmq-server -detached

종료

>> sudo rabbitmqctl stop

2. 샐러리 설치

>> pip install celery

3. 앱 만들기

tasks.py

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

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

'tasks' : 현재 모듈의 이름.

broker='amqp://~' : 사용할 브로커의 URL

4. 샐러리 서버 돌리기

>> celery -A tasks worker --loglevel=info

tasks.py가 들어있는 폴더에 가서 실행해야함.

5. 작업 실행시키기

>> from tasks import add

>> add.delay(4, 4)

task는 실행시켜놓은 워커에 의해 처리됩니다. 아웃풋을 봄으로써 그것을 확인할 수 있습니다.

task를 부르면 AsyncResult 인스턴스를 리턴합니다. 이것은 작업의 상태를 체크하고, 작업이 끝날 때 까지 기다리고, 그것의 리턴값(작업이 실패하면 exception와 traceback)을 받을 때 사용됩니다. 그러나 이것은 디폴트로 작동하지 않아서, 당신은 백엔드를 사용하기 위하여 샐러리를 설정해야합니다. 이것은 다음 세션에 설명되어 있습니다.

6. 결과 저장하기

task들의 상태 변화를 저장하고 싶다면, 샐러리는 어딘가에 상태들을 저장하고 보내야합니다.

사용할 수 있는 선택지들이 몇가지 있습니다. - SQLAlchemy/Django ORM, Memcached, Redis, AMQP (RabbitMQ), MongoDB

그리고 사용자가 정의할 수도 있습니다.

이 예제에서는 rpc 백엔드를 사용할 것입니다. 백엔드는 backend 인자를 통하여(설정 모듈을 사용한다면 CELERY_RESULT_BACKEND 세팅을 통하여) 샐러리에 지정됩니다. 

tasks.py

app = Celery('tasks', backend='rpc://', broker='amqp://guest@localhost//')

로 고침.

설정된 백엔드를 가지고 task를 다시 불러봅시다. 이번에는 task를 부를때 리턴되는 AsyncResult 인스턴스를 저장할 것입니다.

>> from tasks import add

>> result = add.delay(4, 4)

ready() 메소드는 task가 작업을 끝냈는지 아닌지 알려줍니다.

>> result.ready()

False

완료될때까지 결과를 기다려야 합니다. 비동기 콜을 동기로 바꾸는 것이기 때문에 이건 드물게 사용됩니다.

>> result.get(timeout=1)

8

(저는 the operation timed out 에러가 뜨네요.)

task가 예외를 발생시켰을 때, get()은 예외를 다시 발생시킵니다. 그러나 propagate 인자를 정의함으로써 이것을 오버라이드 할 수 있습니다. 

>> result.get(propagate=False)

만약 task가 예외를 발생시키면 오리지널 traceback을 얻을 수 있습니다.

>> result.traceback

7. 환경 설정

샐러리는 소비자 장치와 같이 그렇게 많이 처리 될 필요가 없습니다. 샐러리는 인풋과 아웃풋을 가지고 있고 사용자는 반드시 인풋을 브로커에 연결시켜야 합니다. 만약 원한다면 아웃풋을 백엔드에도요. 그러나 뒤에서 면밀히 살펴본다면 많은 슬라이더, 다이얼, 버튼들을 드러내고있는 덮개가 있습니다. 이게 바로 환경 설정입니다.

기본 환경설정은 많은 용도에 충분합니다만, 수정해야 할 많은 것들이 있습니다. 그러므로 샐러리는 사용자가 원하는 방법으로 사용할 수 있습니다. (설정)가능한 옵션들에 대해 읽어보는 것은 무엇을 설정할 수 있는지 알아보기 위한 좋은 방법입니다. Configuration and defaults 레퍼런스에서 그 옵션들에 대해 읽어볼 수 있습니다.

환경설정은 앱에서 직접 또는 전용 환경설정 모듈을 사용하여 설정할 수 있습니다. 예를 들어서 사용자는 CELERY_TASK_SERIALIZER 세팅을 변경함으로써 연속적인 task payload를 위하여 사용하는 default serializer를 설정할 수 있습니다.

app.conf.CELERY_TASK_SERIALIZER = 'json'

만약 많은 세팅들을 설정한다면 한번에 업데이트할 수 있습니다.

app.conf.update(

    CELERY_TASK_SERIALIZER='json',

    CELERY_ACCEPT_CONTENT=['json'],  # Ignore other content

    CELERY_RESULT_SERIALIZER='json',

    CELERY_TIMEZONE='Europe/Oslo',

    CELERY_ENABLE_UTC=True,

)

큰 프로젝트들에는 전용 환경설정 모듈을 사용하는 것이 유용합니다. 사실 당신(개발자)은 주기적인 task 간격과 task 루팅 옵션들을 하드 코딩 하는 것에 의욕을 잃고는 합니다. 따라서 집중된 위치에 이것들을 모아두는 것이 훨씬 낫습니다. 특히 라이브러리들을 위해서요. 이것은 유저들에게 당신의 task들을 실행시키는 방법을 컨트롤할 수 있도록 합니다. 사용자는 또한 시스템 트러블이 일어났을 때 간단히 환경 설정을 바꾸는 사용자의 SysAdmin을 생각해볼 수 있습니다. // 오역천지

사용자는 환경설정 모듈을 사용하기 위하여 사용자의 샐러리 인스턴스를 부를 수 있습니다. app.config_from_object() 메소드를 이용해서요.

app.config_from_object('celeryconfig')

이 모듈은 보통 celeryconfig라고 불리지만, 아무 이름이나 붙여쓸 수 있어요.

이 모듈의 이름은 celeryconfing.py입니다. 현재 디렉토리 또는 Python 경로에서 불러올 수 있습니다. 이렇게 생겼어요 :

celeryconfig.py:

BROKER_URL = 'amqp://'
CELERY_RESULT_BACKEND = 'rpc://'

CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_TIMEZONE = 'Europe/Oslo'
CELERY_ENABLE_UTC = True

사용자의 환경 설정 파일이 정확히 동작하고 어떠한 구문 오류도 포함하지 않도록, 환경 설정을 적용시키기 위하여 import하세요.

>> python -m celeryconfig

환경 설정 옵션들의 모든 레퍼런스를 보려면 Configuration and defaults를 보세요.

환경 설정의 힘을 보기 위해서, 이건 전용 큐에 예상 외의 task를 보냈을 때 어떻게 할지입니다.

celeryconfig.py:

CELERY_ROUTES = {
    'tasks.add': 'low-priority',
}

또는 이것을 보내는 대신에, task를 제한할 수 있습니다. 그래서 이 타입의 10개의 task들만 1분동안 처리될 수 있습니다.

celeryconfig.py:

CELERY_ANNOTATIONS = {
    'tasks.add': {'rate_limit': '10/m'}
}

만약 브로커로 RabbitMQ 또는 Redis를 사용하고 있다면 워커에 직접 셋할 수 있어요.

>> celery -A tasks control rate_limit tasks.add 10/m

다음 : Next steps