celery 簡單使用

一、簡介
celery是異步任務隊列框架,支持實時處理消息任務，和定時處理消息任務.
架構圖:

組件說明:
producer: 調用celery api，函數，裝飾器，生成task，放入消息隊列的都是生產者
celery beat: 任務調度器,beat進程讀取配置文件，周期性的將配置文件中需要執行的任務放入隊列
broker: 消息代理中間件,task任務存入隊列，供worker取出執行
worker: 執行任務的消費者,可以開啟多個線程執行
backend: 任務處理完后保存狀態信息和結果，以供查詢

二、簡單使用
項目結構

app_test.py實例化celery
from celery import Celery
app = Celery('celerydemo',include=['celerydemo.tasks'])
app.config_from_object('celerydemo.celeryconfig')

if name == 'main':
app.start()

創建任務tasks.py
import time
from celerydemo.app_test import app

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

添加配置文件celeryconfig.py
BROKER_URL = 'redis://:123456@127.0.0.1:6379/0'

CELERY_RESULT_BACKEND = 'redis://:123456@127.0.0.1:6379/0'
CELERY_TASK_SERIALIZER = 'msgpack'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_RESULT_EXPIRES = 60 60 24
CELERY_ACCEPT_CONTENT = ['json', 'msgpack']

創建worker消費者線程
celery -A celerydemo.app_test worker -l info

調用任務exec.py 放入隊列,并輸出結果
from celerydemo.tasks import add
import time
t1 = time.time()

#調用delay將任務插入broker隊列
r1 = add.delay(1,2)
r2 = add.delay(2,2)
r3 = add.delay(3,2)
r4 = add.delay(4,2)
r5 = add.delay(5,2)

r_list = [r1,r2,r3,r4,r5]

for r in r_list:
while not r.ready():
pass
print(r.result)

t2 = time.time()

print('共耗時: %s' % str(t2-t1))