創建計划任務：
from celery import Celery import time my_task=Celery("task",broker="redis://:123456@127.0.0.1:6379",backend="redis://:123456@127.0.0.1:6379") @my_task.task def my_func1(): time.sleep(10) return "任務1" @my_task.task def my_func2(): return "任務2" @my_task.task def my_func3(): return "任務"
調用方法執行指定的任務：
from s1 import my_func1 res=my_func1.delay() print(res)
獲取返回值中運行計划的ID
判斷計划是否執行完成：
from celery.result import AsyncResult from s1 import my_task async_task=AsyncResult(id="48029f4f-769e-438b-ac97-e89cc0bb1157",app=my_task) # result=async_task.get()
if async_task.successful(): result=async_task.get() print(result+"OK!") else: print("任務還未執行完成！")
啟動celery在命令行執行： Celery worker -A s1 -l INFO -P eventlet -c 6
-A:指定要執行的目錄
-l: 指定要使用的打印日志級別
-p:指定使用eventlet插件 讓高版本celery支持window平台
-c:指定可執行的計划數量