Celery from scratch

Celery Startup

Celery Coding

Coding은 매우 간단하다.(아래 샘플)

 

[file 명 : tasks.py]

from celery import Celery

app = Celery('tasks', broker="broker url", backend="backend url")

@app.task

def add(x, y):

    return x+y

 

이렇게 하면 기본적인 코딩 끝이다.

@app.task는 celery에 task를 등록한다는 것이다.

 

Celery 실행

실행방법

celery config paramters

 

http://docs.celeryproject.org/en/latest/userguide/configuration.html#example-configuration-file

ex) CELERYD_CONCURRENCY 는 celery 의 동시 실행 worker갯수를 의미한다.

따라서, 해당 변수를 setting하면 제어가능하다.

위의 변수를 setting안하면 default는 12이다.

따라서 아래 명령어 celery 를 실행하고 ps -eaf | grep celery | grep -v grep 하면 총 13개가 나온다.

1개는 master 이며 나머지 12개가 worker이다.

app = Celery(...), app.conf.update(CELERYD_CONCURRENCY=3) 이렇게 하고 다시 ps 해보면 4가 나온다.

 

celery -A tasks worker --loglevel=info

-A: application ( = -app )  => 확장자를 제외하고 입력함. 만약 tasks.py 이면 tasks를 입력

worker 는 기본 명령어이며 worker 서버를 구동하라는 뜻이다.

--loglevel은 다들 아는 것이니 넘어가자.

정리 : -A 파이썬화일명(확장자제외) 이부분만 기억하면 된다. 

 

celery worker 에 작업시키기

python prompt로 들어감.

중요한 것은, @app.task를 통해 등록한 add task를 호출하는 것은 delay 이다.

delay : task call

위에 key가 redis에 저장되는 키이다. ( celery-task-meta-  + key )

위와 같이 redis-cli로 들어가서 get key정보를 하면 값이 나온다. result를 보면 6이다.

task_id는 key가 된다.

 

비동기 작업 수행

Celery 활용시 필수 사항

대부분 비동기처리를 위해 Celery를 사용한다.

그리고 서버는 일반적으로 여러대이며 어떤 서버(그리고 어떤 Celery데몬이) 가 해당 Task를 수행할지 모른다.

예를 들면 File Upload 및 데이터 처리시 1번서버에서 Upload를 하면 1번서버에서 수행되어야 하지만,

2번서버의 Celery데몬이 작업을 수행하면 File not found error가 난다.

또는 2개의 데몬이 있는데 대량 작업이 돈다면 그 작업은 하나의 데몬, 나머지 작업은 다른 데몬이 하도록 하게 하고 싶은 경우,

Queue라는 개념이 매우 중요하다.

 

예를 들면,

add 라는 task가 있다고 하자.

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

위 Task는 30초 걸리는 장시간 작업이다.

celery -A proj worker -l info
celery -A proj worker -l info -Q hipri

위 데몬에  -Q option이 큐지정이다.

-Q option : 기본값은 celery이다. 즉 없는 것은 -Q celery와 같다.

-Q hipri : queue가 hipri로 들어오면 수행하는 데몬이고 나머지는 수행안한다.

apply_async를 써야 real-time으로 queue를 제어할 수 있고, delay를 사용할려면 queue는 속성으로 한번 지정되면,

다시 수정해서 celery데몬을 띄워야 하니 나는 항상 apply_async를 사용한다.

1) add.apply_async((2,5))
2) add.apply_async((2,3), queue='celery')
3) add.apply_async((2,5), queue='hipri')

1)번과 2)번은 default로 동일하다.

3)번은 데몬의 처리queue가 hipri인 대상이 수행하라는 뜻이다.

 

이걸로 아래의 실무에 적용할수 있다.(예시임.)

sequenced : long running job은 하나의 데몬이 순차적으로 수행

paralleled : 가벼운 작업은 여러 데몬이 병렬로 수행

near_real : 준실시간 처리는 여기로... 등등

 

여러 celery데몬을 띄워서 제어하는 것 까지 실무에서 많이 사용한다.

이상 기본적인 Celery를 마쳤으며 이정도만 알아도 활용할 수 있다.

더 고급정보는 실무에 활용하면서 배우기 바란다.