celery包结构的使用
包结构如下
项目名
├── celery_task # celery包
│ ├── __init__.py # 包文件
│ ├── celery.py # celery连接和配置相关文件，且名字必须叫celery.py app文件必须在这里面创建出来
│ └── user_tasks.py # 所有任务函数
└── order_tasks.py # 所有任务函数
├── add_task.py # 添加任务
└── get_result.py # 获取结果
1 新建一个包下面存放如下的文件
celery_demo2
| celery_task
│ celery.py # 必须叫这个名称 celery.py ！！！
│ order_task.py
│ user_task.py
│ __init__.py # 注意 必须是一个包
| add_task.py
| get_result.py
2 celery.py
from celery import Celery
broker = "redis://127.0.0.1:6379/1"
backend = "redis://127.0.0.1:6379/2"
# 监控任务列表里面放模块，不需要加.py 会自动在app里面注册
app = Celery("app", broker=broker, backend=backend, include=["celery_task.order_task", "celery_task.user_task"])
3 order_task.py
import time
from .celery import app
@app.task
def send_sms(mobile):
time.sleep(3)
return (f"{mobile} 下单成功")
4 user_task.py
import time
from .celery import app
@app.task
def add(x, y):
time.sleep(3)
return x + y
5 add_task.py 有坑 -->取id
from celery_task.order_task import send_sms
# 同步调用
# send_sms('187xxxx2008')
# 异步调用
res = send_sms.delay("187xxxx2008")
print(res.id) # 这里必须取id!!!!!!
6 get_result.py
from celery_task.celery import app
from celery.result import AsyncResult
id = '38429ddf-c0f1-4f83-b8f1-5f0c71b8e129'
if __name__ == '__main__':
result = AsyncResult(id=id, app=app)
if result.successful():
result = result.get()
print(result)
elif result.failed():
print('任务失败')
elif result.status == 'PENDING':
print('任务等待中被执行')
elif result.status == 'RETRY':
print('任务异常后正在重试')
elif result.status == 'STARTED':
print('任务已经开始被执行')
7 依次执行拿结果 有坑 --> 导入到包就行
add_task.py
celery -A celery_task worker -l info -P eventlet # 注意 这里导入到包这一层就可以了 celery_task
get_result.py