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))