农业网站模板免费下载,网站建设html5源码,AAP网站开发需要多少钱,前端入职后很痛苦Celery的任务流
在之前调用任务的时候只是使用delay()和apply_async()方法。但是有时我们并不想简单的执行单个异步任务#xff0c;比如说需要将某个异步任务的结果作为另一个异步任务的参数或者需要将多个异步任务并行执行#xff0c;返回一组返回值#xff0c;为了实现此…
Celery的任务流
在之前调用任务的时候只是使用delay()和apply_async()方法。但是有时我们并不想简单的执行单个异步任务比如说需要将某个异步任务的结果作为另一个异步任务的参数或者需要将多个异步任务并行执行返回一组返回值为了实现此目标Celery使用一种叫做signatures的东西
celery的简单使用
signature的引入
signature官方文档
可以简单理解为signature是将之前的异步任务以某种方式包装包装后的异步任务仍可以使用之前的delay()和apply_async()方法并且包装后的异步任务就可以以多种方式组合成复杂的工作流程
先创建一个tasks.py
import timeimport celerybroker redis://127.0.0.1:6379/1
backend redis://127.0.0.1:6379/2
app celery.Celery(app, backendbackend, brokerbroker)# 加
app.task
def add_num(a, b):print(f{a}{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a breturn c# 减
app.task
def subtract_num(a, b):print(f{a}-{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a - breturn c# 乘
app.task
def multiply_num(a, b):print(f{a}*{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a * breturn c# 除
app.task
def divide_num(a, b):print(f{a}/{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a / breturn capp.task
def test(args):print(args)return args
运行celery消费者
# 格式为celery -A app对象所在的文件 worker -l 日志级别 -Q 队列名称(也可以不指定默认为celry)
celery -A tasks worker -l info -Q test用signature对上面的add_num包装
from celery import signature
from tasks import add_num, subtract_num, multiply_num, divide_num# 方法1
sign add_num.signature((1, 1), queuetest)
ret sign.delay()
print(ret.get())# 方法2
sign signature(tasks.add_num, (1, 1), queuetest)
ret sign.delay()
print(ret.get())# 方法3
sign signature(add_num, (1, 1), queuetest)
ret sign.delay()
print(ret.get())
chain的使用
chain官方链接
chain可以将signature包装的任务函数一个一个执行一个执行完将执行return结果传递给下一个任务函数
from celery import signature, chain
from tasks import add_num, subtract_num, multiply_num, divide_numadd_sign signature(add_num, (6, 2), queuetest)
subtract_sign signature(subtract_num, (2,), queuetest)
multiply_sign signature(multiply_num, (2,), queuetest)
divide_sign signature(divide_num, (2,), queuetest)# 对某个数依次做加减乘除处理
chain1 chain(add_sign, subtract_sign, multiply_sign, divide_sign)
ret chain1.delay()
print(ret.get())可以看到异步任务依次执行并将上一个异步任务的结果作为参数传递给下一个形成一个链条
group的使用
group官方链接
group可以将signature包装的任务函数并行执行返回一组返回值
from celery import signature, chain, group
from tasks import add_num, subtract_num, multiply_num, divide_numadd_sign signature(add_num, (6, 2), queuetest)
subtract_sign signature(subtract_num, (6, 2), queuetest)
multiply_sign signature(multiply_num, (6, 2), queuetest)
divide_sign signature(divide_num, (6, 2), queuetest)# 对某个数分别做加减乘除处理group1 group(add_sign, subtract_sign, multiply_sign, divide_sign)
ret group1.delay()
print(ret.get())
#[8, 4, 12, 3.0]可以看到相比于chaingroup里的任务是同时执行
chord的使用
chord官方链接 依赖一个group任务group任务结束后将所有子任务的返回值作为参数传递给chord的回调函数即chord由group任务组与回调函数组成
上代码
from celery import signature, chain, group, chord
from tasks import add_num, subtract_num, multiply_num, divide_num, testadd_sign signature(add_num, (6, 2), queuetest)
subtract_sign signature(subtract_num, (6, 2), queuetest)
multiply_sign signature(multiply_num, (6, 2), queuetest)
divide_sign signature(divide_num, (6, 2), queuetest)group1 group(add_sign, subtract_sign, multiply_sign, divide_sign)#包装test异步任务函数
test_sign signature(test, queuetest)c1 chord(group1, test_sign)
c1.delay()可以看出在执行完加减乘除所有异步任务后chord会将任务组的结果作为list交给test函数这里的test有点像回调函数
PS根据我的观察chaingroupchord在执行完后都会返回一个任务id其中chain的任务id为任务链里最后一个任务的idgroup的任务id是一个临时的任务idgroup任务都结束后就会消失chord的任务id是回调函数的任务id。因此chain和chord在任务结束后任务结果还是可以查到的而group则查询不到因此group的任务结果可能无法用AsyncResult查询到
最后附上celery关于任务工作流的官方链接 celery工作流
PS
有的时候我们可能需要在celery的task函数中调用其他的celery函数并且需要同步的获取结果其实着本质上就是把异步的celery函数变成同步运行具体如下先创建一个tasks.py
import timeimport celerybroker redis://127.0.0.1:6379/1
backend redis://127.0.0.1:6379/2
app celery.Celery(app, backendbackend, brokerbroker)# 加
app.task
def add_num(a, b):print(f{a}{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a breturn capp.task
def test():#在test函数里调用add_num函数并且同步获取结果,将结果作为test函数的返回值ret add_num.delay(1,2)ret ret.get()return ret#启动消费者
celery -A tasks worker -l info调用test异步函数
from tasks import test
ret test.delay()结果就是出错了因为官方不建议在一个异步任务中区等待另一个异步任务的返回结果所以这个时候就可以通过上面的chain方法实现这个需求。当然还有一种不建议的方法就是在同步获取celery任务结果的get方法中添加参数disable_sync_subtasksFalse具体如下
import timeimport celerybroker redis://127.0.0.1:6379/1
backend redis://127.0.0.1:6379/2
app celery.Celery(app, backendbackend, brokerbroker)# 加
app.task
def add_num(a, b):print(f{a}{b})time.sleep(3) # 做延时处理方便后面查看任务执行顺序c a breturn capp.task
def test():ret add_num.delay(1, 2)ret ret.get(disable_sync_subtasksFalse)#在这添加disable_sync_subtasksFalsereturn ret再调用一次test方法 成功调用
详见celery官方链接 链接传送门
结语
写这些仅记录自己学习使用celery的过程。如果有什么错误的地方还请大家批评指正。最后希望小伙伴们都能有所收获。
