Asynchronous Tasks with Django and Celery(web에서 최적의 구현 방법을 보여준다)

Celery의 Subtask 기능을 이용하여 Chord와 Chain로 작업 분산해서 다루기

아래와 같이 task1은 100초 걸리는 작업, task2는 바로 리턴하는 작업을 만들었다. beat_schedule 에는 task1을 10초마다 호출, task2를 1초마다 호출했다.

@app.task
def task1():
    import time
    time.sleep(100)
    return "task1"

@app.task
def task2():
    return "task2"

app.conf.beat_schedule = {
    'task1': {
        'task': 'aip.celery.task1',
        'schedule': 10,
        'options': {'queue': 'task1'},
    },
    'task2': {
        'task': 'aip.celery.task2',
        'schedule': 1,
        'options': {'queue': 'task1'},
    },
}

그러자 아래와 같이 queue에 task1때문에 작업이 어느정도 쌓이다가 더 이상 쌓이지 않게 되었다.

어디선가 task1을 수행하느라 걸린 것 같다. task1이 끝나자마자 아래와 같이 무수히 많은 task들이 진행되었다.

이대로는 안된다. 처리 속도보다 쌓이는 속도가 더 많아서 과부하가 걸린다.

apply_async 호출 시 task_id를 어딘가에 저장하고, 나중에 결과만 받아오도록 설정하자.

# apply_async 호출하는 부분
result = task1.apply_async(queue="task1")

# task_id를 저장하는 부분
status_check_service.set_task_id(result.task_id)

# 결과를 받아오는 부분
data = status_check_service.get_task_queue_status()

주기적으로 호출하는 task인데 오래걸리는 task라면 이전에 아직 수행중인 작업이 남아 있는지 검사하자

• apply_async로 호출 당하는 함수는 함수 내에 return을 해주어야 result가 저장이 된다
  • return을 해줘도 다음으로 넘어갈 수 있음

# 호출 당하는 함수
@app.task
def test_sleep():
    import time
    time.sleep(10)
    return {
        "task_status": consts.CELERY_STATUS_SUCCESS,
    }

# 호출하는 함수
@app.task
def test1():
    if queue_check_service.get_status() in [consts.CELERY_STATUS_SUCCESS, consts.CELERY_STATUS_FAILURE]:
        queue_info = test_sleep.apply_async(queue="test1")
        queue_check_service.set_task_id(queue_info.task_id)
    return {
        "task_status": consts.CELERY_STATUS_PENDING,
    }