1、首先两个文件
<1> tasks （一般名称固定）
tasks.py
from celery import Celery
import time
# app = Celery('tasks')
app = Celery('celery_name',
backend='redis://127.0.0.1:6379/3',
broker='redis://127.0.0.1:6379/2') # 配置好celery的backend和broker
# app.config_from_object('tasks') #config是本目录下的config.py文件名,指定自己的目录即可
@app.task(name='func') # 这里需要注册任务, name参数的值是任务名称，注意是字符串
def func(x, y):
time.sleep(3)
print('已经运行。。')
return x + y
<2>第二个文件是异步任务调度的入口?
celery_test_main.py
from tasks import func
import time
for i in range(100):
func.delay(2, i)
time.sleep(1)2、启动异步
celery -A tasks worker --loglevel=info
3、运行celery_test_main.py文件添加任务
这里可以看到redis 2 号库是这样的
?其实任务主要存在celery里面，这个key是个列表
?列表里面有很多任务，里面的某一个元素是字符串的这种数据结构：
{
"body": "W1syLCA5OV0sIHt9LCB7ImNhbGxiYWNrcyI6IG51bGwsICJlcnJiYWNrcyI6IG51bGwsICJjaGFpbiI6IG51bGwsICJjaG9yZCI6IG51bGx9XQ==",
"content-encoding": "utf-8",
"content-type": "application/json",
"headers": {
"lang": "py",
"task": "func",
"id": "3fb1edf9-9b4e-4671-a1a4-c9a23b68a2b8",
"shadow": null,
"eta": null,
"expires": null,
"group": null,
"group_index": null,
"retries": 0,
"timelimit": [null, null],
"root_id": "3fb1edf9-9b4e-4671-a1a4-c9a23b68a2b8",
"parent_id": null,
"argsrepr": "(2, 99)",
"kwargsrepr": "{}",
"origin": "gen15587@iZbp1j4bsah4npvyegtlioZ",
"ignore_result": false
},
"properties": {
"correlation_id": "3fb1edf9-9b4e-4671-a1a4-c9a23b68a2b8",
"reply_to": "3bfd1eca-d23d-3bfe-8ce9-29f2dcbc27be",
"delivery_mode": 2,
"delivery_info": {
"exchange": "",
"routing_key": "celery"
},
"priority": 0,
"body_encoding": "base64",
"delivery_tag": "46650cc0-690c-43f1-b97d-6881ba9b358f"
}
}
4、当任务都执行完毕之后，这个celery的列表就不存在了
5、而如果你把celery服务停止之后，运行celery_test_main.py文件，依然是能在redis的2号库添加任务；可以看到任务堆积了
而任务貌似是不会过期的:
?6、重启异步可以看到仍然可以正常执行未执行的任务
7、测试用3号库当作 backend 存储任务结果的库，里面结构是这样的
可以看到任务执行的成功与否的状态和执行完成的时间；以及是哪个任务(task_id)
8、可以再开一个终端再以同样的命令启动一个worker
可以看到启动两个更快，而且分别执行不同的任务(任务不会重复执行，取到哪个执行哪个worker从队列中取到哪个执行哪个）
9、如果任务执行失败看看是怎样的？
<1>修改下代码，故意出错
修改之后依然能成功执行，我纳闷了，是不是该重启异步？重启异步试试
重启一个异步之后，只有重启的这个会失败，另外一个没有重启的异步还能成功：
?
?结果库：
有失败的，有成功的，看来修改完异步任务的话，要重启worker的，不然不生效啊