flask+celery处理异步任务
celery是一个强大的分布式任务队列，在这里我们介绍一下它最基本的处理异步任务的功能，包含以下几个：
-
创建Celery实例
-
创建一个异步任务
-
查询异步任务的信息
-
取消异步任务
使用的环境是flask3.0+Celery5.4
1. 创建Celery实例
celery通过中间件获取任务，再将任务结果写入backend，所以创建实例时需要传递这两个的url，其支持多种协议（比如redis、rabbitmq等），这里我们以redis为例。最简单的一种是创建对象时传递：
from celery import Celery
celery_app = Celery(__name__, broker='redis://localhost:6379/0', backend='redis://localhost:6379/1')
另外一种是使用配置文件，按官方文档的例子，在同级目录创建一个celeryconfig.py文件，写入配置信息，然后更新celery对象的配置：
celery_app = Celery(__name__)
apcelery_appp.config_from_object('celeryconfig')
但可能是由于Celery版本问题，配置字段改变了（大小写改变），这种方法连接backend一直报错，暂时还没找到原因
2. 创建一个异步任务
类似于flask，celery也使用装饰器来创建任务处理函数，比如官方例子：
@celery_app.task
def add(x, y):
return x + y
但这种方式无法访问任务实例的信息，需要修改一下装饰器的参数，将任务实例作为第一个参数引入：
@celery_app.task(bind=True)
def add(self, x, y):
print(self.request.id) # 打印任务
return x + y
返回值会写到backend，存储在result字段。结合flask的例子如下：
from flask import Flask
# 其它代码
# ...
app = Flask(__name__)
@app.route('/task', methods=['POST'])
def add_task():
id = add.delay(1, 2) # 返回的是异步任务的
return {'code': 0, 'data': id}
注意：Celery是要单独启动的，然后再启动flask，启动命令类似于：
Celery -A app worker --loglevel=info
这个地方也有比较多的坑，经常报模块不存在这样的错
3. 查询异步任务的信息
大部分异步任务的耗时都比较长，过程中用户可能想知道任务执行的情况，比如进度等，这时前面引入的第一个参数（self）就能发挥作用了，我们修改一下demo， 每隔1秒计算一次，并将进度更新到backend：
import time
@celery_app.task(bind=True)
def add(self, x, y):
for i in range(1, 100):
print(x + y)
self.update_state(state="PROGRESS", meta={'index': i})
time.sleep(1)
return x + y
而对应的flask查询代码如下：
from celery.result import AsyncResult
@app.route('/task/<string:id>', methods=['GET'])
def get_task(id):
return {'code': 0, 'data': celery_app.AsyncResult(id)}
4. 取消异步任务
取消任务要借助Celery的Control类：
@app.route('/task/<string:id>', methods=['DELETE'])
def remove_task(id):
ctrl = Control(celery_app)
ctrl.revoke(id)
return {'code': 0}
原文地址：https://blog.csdn.net/linqiaqun/article/details/143420784
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！