版權宣告：本文為博主原創文章，未經博主允許不得轉載。https://mp.csdn.net/mdeditor/83016539

任務傳送者，任務執行者，中間人

安裝: pip install celery

使用:
1）建立一個Celery類的例項物件
from celery import Celery
celery_app = Celery('demo')

2）進行配置，配置broker的地址，可以新建一個配置檔案
配置檔案：broker_url = '中間人地址'

3）載入配置
celery_app.config_from_object('配置檔案路徑')

4）封裝任務函式
@celery_app.
 
task(name='my_first_task')
def task_func(a, b):
    print('任務函式被呼叫')
    # ...
 
5）建立worker
celery -A `celery_app物件包路徑` worker -l info

6）發出要執行任務
task_func.delay(1, 3)

建立一個資料夾celery_tasks

• main.py檔案來進行我們celery執行的入口

"celery_tasks 資料夾下 main.py檔案"

# 建立celery類的物件
from celery import Celery
celery_app =
 
 Celery('celery_tasks')

# 載入配置項
celery_app.config_from_object("celery_tasks.config")

# 啟動worker時 自動發現 任務函式
"worker發現任務函式的時候自動去路徑下找tasks檔案，所以任務函式檔名一定要是tasks"
celery_app.autodiscover_tasks(['celery_tasks.sms'])

• 資料夾下建立config.py檔案，用來儲存celery配置資訊

# 指定中間人的地址
broker_url = "redis://127.0.0.1:6379/3"
 

• 封裝任務函式-- 因為後期會有很多的任務函式，所以我們為任務建立資料夾

# 建立一個發簡訊的任務，建立資料夾 sms 資料夾儲存發簡訊的函式檔案
# 建立一個tasks.py檔案，用來寫發簡訊的函式
"tasks.py檔案"
from celery_tasks.main import celery_app

@celety_app.tasks(name="send_sms_code")
def send_sms_code(a, b):
	print("任務函式被呼叫了")

• 開啟我們的celery

celery -A celery_tasks.main worker -l info

• 發出我們要執行的任務

"view.py"
from celery_tasks.sms.tasks import send_sms_code

send_sms_code.delay("要傳遞的引數")