[Celery] Celery with Redis (python) : 비동기 작업 queue

✅ Celery ?

celery는 분산 메시지 전달에 기반을 둔 오픈 소스 비동기 테스크 큐, 잡 큐이다.

스케줄링을 지원하지만 실시간 운영에 초점을 두고있다.

 

1️⃣ 시작

redis가 실행되어 있어야 한다. (아래 링크 참고)

[Redis] redis를 시작하는 간단한 방법 2가지 (docker/source download)

 

2️⃣ 설치

pip3 install celery

 

3️⃣ 실행

tasks.py

 

from celery import Celery

app = Celery('tasks', broker = 'redis://localhost//')

app.conf.task_default_queue = 'default'
app.conf.task_queues = (
    Queue('default', routing_key='task.#'),
)

@app.task
def long_task(x):
    print(f"{x} : start")
    pytime.sleep(10)
    print(f"{x} : end")
    return x

 

 

terminal 에서 celery worker server 실행

 

python3 -m celery -A tasks worker -Q default --loglevel=info

 

 -------------- celery@MacBook-Pro.local v5.2.6 (dawn-chorus)
--- ***** ----- 
-- ******* ---- macOS-10.16-x86_64-i386-64bit 2022-05-30 09:33:30
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7fef50fb9550
- ** ---------- .> transport:   redis://localhost:6379
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> default          exchange=default(direct) key=task.#
                

[tasks]
  . tasks.long_task

[2022-05-30 09:33:30,586: INFO/MainProcess] Connected to redis://localhost:6379/0
[2022-05-30 09:33:30,588: INFO/MainProcess] mingle: searching for neighbors
[2022-05-30 09:33:31,598: INFO/MainProcess] mingle: all alone
[2022-05-30 09:33:31,620: INFO/MainProcess] celery@MacBook-Pro.local ready.

 

1. app : 현재 서버로 가동한 app의 이름과 ID

2. transport : message broker

3. result : task 수행 결과를 보낼 backend (일반적으로 celery는 result backend를 지정하지 않는다.)

4. concurrency : prework된 worker 프로세스 숫자 (일반적으로 CPU 코어 숫자만큼 가동된다.)

 

 

4️⃣ 동시성(concurrency, worker) 지정하기

이번에 진행할 작업이 긴 시간이 소요되는 작업이다.

요청이 들어오면 queue 방식으로 다 쌓아두고 FIFO으로 하나씩 처리하려한다.

 

그러기 위해서는 동시성을 1로 지정해주기 위해서는 celery 서버를 실행시킬때 `--concurrency=1` 로 지정해준다.

 

python3 -m celery -A tasks worker -Q default --loglevel=info --concurrency=1

 

 -------------- celery@MacBook-Pro.local v5.2.6 (dawn-chorus)
--- ***** ----- 
-- ******* ---- macOS-10.16-x86_64-i386-64bit 2022-05-30 09:32:44
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7f97a1069a00
- ** ---------- .> transport:   redis://localhost:6379
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 1 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> default          exchange=default(direct) key=task.#
                

[tasks]
  . tasks.long_task

[2022-05-30 09:32:45,095: INFO/MainProcess] Connected to redis://localhost:6379/0
[2022-05-30 09:32:45,097: INFO/MainProcess] mingle: searching for neighbors
[2022-05-30 09:32:46,105: INFO/MainProcess] mingle: all alone
[2022-05-30 09:32:46,123: INFO/MainProcess] celery@MacBook-Pro.local ready.

 

 

☑️ celery worker guide

Workers Guide — Celery 5.2.7 documentation

 

5️⃣ 테스트

실행 코드

 

for i in range(1, 6):
    long_task.apply_async(args=[i], queue='default')

 

실행 결과

 

[2022-05-30 09:44:24,594: INFO/MainProcess] Task tasks.long_task[e97349bd-1807-4019-906c-f61e10d3cbc1] received
[2022-05-30 09:44:24,596: INFO/MainProcess] Task tasks.long_task[a3d4ed9d-40b9-4a95-9b22-28c24f835b04] received
[2022-05-30 09:44:24,596: WARNING/ForkPoolWorker-1] 1 : start
[2022-05-30 09:44:24,597: INFO/MainProcess] Task tasks.long_task[6395d77d-708f-441e-8a93-fa47726f9b01] received
[2022-05-30 09:44:24,597: INFO/MainProcess] Task tasks.long_task[bfeec24d-96ba-4560-99b2-18faec595e03] received
[2022-05-30 09:44:24,598: INFO/MainProcess] Task tasks.long_task[6992e54a-0e86-4582-bbb4-4b9d66829a53] received
[2022-05-30 09:44:34,597: WARNING/ForkPoolWorker-1] 1 : end
[2022-05-30 09:44:34,600: INFO/ForkPoolWorker-1] Task tasks.long_task[e97349bd-1807-4019-906c-f61e10d3cbc1] succeeded in 10.004834291999998s: 1
[2022-05-30 09:44:34,605: WARNING/ForkPoolWorker-1] 2 : start
[2022-05-30 09:44:44,606: WARNING/ForkPoolWorker-1] 2 : end
[2022-05-30 09:44:44,608: INFO/ForkPoolWorker-1] Task tasks.long_task[a3d4ed9d-40b9-4a95-9b22-28c24f835b04] succeeded in 10.004181875000002s: 2
[2022-05-30 09:44:44,611: WARNING/ForkPoolWorker-1] 3 : start
[2022-05-30 09:44:54,613: WARNING/ForkPoolWorker-1] 3 : end
[2022-05-30 09:44:54,615: INFO/ForkPoolWorker-1] Task tasks.long_task[6395d77d-708f-441e-8a93-fa47726f9b01] succeeded in 10.004144500000002s: 3
[2022-05-30 09:44:54,618: WARNING/ForkPoolWorker-1] 4 : start
[2022-05-30 09:45:04,619: WARNING/ForkPoolWorker-1] 4 : end
[2022-05-30 09:45:04,620: INFO/ForkPoolWorker-1] Task tasks.long_task[bfeec24d-96ba-4560-99b2-18faec595e03] succeeded in 10.002915666s: 4
[2022-05-30 09:45:04,621: WARNING/ForkPoolWorker-1] 5 : start
[2022-05-30 09:45:14,622: WARNING/ForkPoolWorker-1] 5 : end
[2022-05-30 09:45:14,623: INFO/ForkPoolWorker-1] Task tasks.long_task[6992e54a-0e86-4582-bbb4-4b9d66829a53] succeeded in 10.003116958s: 5

 

 

 

☑️ celery routing에 대한 설정 값 관련 guide

Routing Tasks — Celery 5.2.7 documentation