Celery框架從入門到精通
2023-03-09 21:01:04
目錄
Celery介紹、安裝、基本使用
一、Celery服務
什麼是Celery:
Celery是一個簡單、靈活且可靠的，處理訊息的分散式系統
- Celery可以用來做什麼：
- 非同步任務
- 定時任務
- 延遲任務
Celery的執行原理:
- 可以不依賴任何服務，通過自身命令，啟動服務
- celery服務為其他專案服務提供非同步解決任務需求
# 注：會有兩個服務同時執行
- 專案服務
- celery服務
專案服務將需要非同步處理的任務交給celery服務，celery就會在需要時非同步完成專案的需求
'''
人是一個獨立執行的服務 | 醫院也是一個獨立執行的服務
正常情況下，人可以完成所有健康情況的動作，不需要醫院的參與；但當人生病時，就會被醫院接收，解決人生病問題
人生病的處理方案交給醫院來解決，所有人不生病時，醫院獨立執行，人生病時，醫院就來解決人生病的需求
'''
1、celery架構
-
訊息中介軟體：broker
- 提交的任務【函數】都放在這裡， celery本身不能提供訊息中介軟體
- 需要藉助於第三方： redis或rabbitmq
-
任務執行單元：worker
- 真正執行任務的的地方，一個個程序中執行函數
-
結果儲存：backend
- 函數return的結果都儲存在這裡， celery本身不提供結果儲存
- 需要藉助於第三方: redis或rabbitmq
使用場景：
- 非同步執行：解決耗時任務
- 延遲執行：解決延遲任務
- 定時執行：解決週期任務
2、celery快速使用
Celery不支援在windows上直接執行，通過eventlet支援在win上執行
安裝：
pip install celery
pip install eventlet # windows需要安裝
快速使用：
- 1、第一步：建立一個py檔案（main.py），用於範例化celery物件，編寫需要執行的函數
# 1、匯入模組
from celery import Celery
# 2、指定briker，用於存放提交的非同步任務
broker = 'redis://127.0.0.1:6379/1'
# 3、指定backend，用於存放函數執行結束的結果
backend = 'redis://127.0.0.1:6379/2'
# 範例化celery物件
app = Celery('test', broker=broker, backend=backend)
# 編寫一個函數，裝飾上celery物件
@app.task
def add(a, b):
import time
time.sleep(3)
print('add函數執行完成')
return a + b
- 2、第二步：再次建立一個py檔案（run.py）,用於將函數提交給celery
# 1、匯入剛才編寫的函數
from main import add
# 2、將任務提交給broker,函數需要的引數需要傳入
res = add.delay(1, 2)
# 3、提交後可以獲得該任務的ID，可通過ID可以查詢任務執行結果
print(res) # 0213d2c2-453e-41a8-a171-e31f1f2f4883
- 3、第三步：使用命令開啟worker （也可以提前開啟，任務提交後就會直接執行）
# 啟動worker命令，win需要安裝eventlet
# 啟動需要進入main.py檔案的目錄下
win:
-4.x之前版本
celery worker -A main -l info -P eventlet
-4.x之後
celery -A main worker -l info -P eventlet
mac：
celery -A main worker -l info
- 4、第四步：worker會將執行的結果存在之前指定的broker目錄下（指定的redis資料庫）
- 5、第五步：通過程式碼檢視執行結果（建立新的py檔案，專門用於檢視執行結果）
# 1、匯入celery範例的物件
from main import app
# 2、匯入該模組用於檢視結果
from celery.result import AsyncResult
# 3、將提交的任務編號拿過來，用於查詢結果
id = '0213d2c2-453e-41a8-a171-e31f1f2f4883'
# 4、指定該檔案為啟動檔案
if __name__ == '__main__':
# 範例化物件，將任務的ID和celery範例化物件當作引數傳入
a = AsyncResult(id=id, app=app)
# 判斷執行結果
if a.successful(): # 執行完了
result = a.get()
print(result)
elif a.failed():
print('任務失敗')
elif a.status == 'PENDING':
print('任務等待中被執行')
elif a.status == 'RETRY':
print('任務異常後正在重試')
elif a.status == 'STARTED':
print('任務已經開始被執行')
二、Celer包結構
1、建立clery包結構
什麼是包結構：通過將celery服務封裝成包的形式，放在專案需要使用的時候匯入即可
project
├── celery_task # celery包
│ ├── __init__.py # 包檔案
│ ├── celery.py # celery連線和設定相關檔案，且名字必須交celery.py
│ └── tasks.py # 所有任務函數
├── add_task.py # 新增任務
└── get_result.py # 獲取結果
建立包：
建立一個包，名為：celery_task
- 1、第一步：在包下建立py檔案（名字必須為celery.py）
# 匯入celery模組
from celery import Celery
# 匯入設定broker和backend
from .settings import BACKEND, BROKER
# 範例化celery物件
app = Celery('test',
broker=BROKER,
backend=BACKEND,
include=['celery_task.order_task',
'celery_task.user_task'])
- 2、第二步：建立settings.py，用於存放設定
BROKER = 'redis://127.0.0.1:6379/1'
BACKEND = 'redis://127.0.0.1:6379/2'
- 3、第三步，建立py檔案（task.py），用於存放需要執行的非同步任務
# 匯入celery範例物件
from .celery import app
# 計算函數
@app.task()
def add(a, b):
print('計算結果為:', a + b)
return True
# 模擬傳送簡訊
@app.task()
def send_sms(mobile, code):
print('已向手機號：%s 傳送簡訊，驗證碼為：%s' % (mobile, code))
return True
- 4、第四步：開啟worker
切換到celery所在的目錄下，開啟worker命令
celery -A celery_task worker -l info -P eventlet
- 5、第五步：提橋任務： # add_task.py 檔案下
# 提交任務，這裡模擬的是非同步任務的提交
res = add.delay(a, b) # 提交後可以接收任務的ID
res1 = send_sms.delay(mobile, code)
- 6、第六步：檢視任務執行結果： # get_result.py 檔案下
# 匯入celery範例
from celery_task.celery import app
from celery.result import AsyncResult
id = res
id1 = res1
# 通過傳入任務的ID就可以查詢到任務的執行結果
def res_func(id):
id = id
a = AsyncResult(id=id, app=app)
if a.successful(): # 執行完了
result = a.get()
if result: return '執行完成'
elif a.failed():
return '任務失敗,失敗的原因可能是未開啟worker'
elif a.status == 'PENDING':
return '任務等待中被執行，當前任務較多或未開啟worker'
elif a.status == 'RETRY':
return '任務異常後正在重試'
elif a.status == 'STARTED':
return '任務已經開始被執行，請稍後查詢'
2、Celery執行非同步任務、延遲任務、定時任務
執行非同步任務：
# 程式碼用法：
函數名.delay('函數執行需要的引數')
res = func.delay(*args,**kwargs) # res 用於接收提交任務的ID
執行延遲任務：
# 程式碼用法：
# 1、執行延遲任務
from datetime import datetime, timedelta
# 設定延遲後的時間，一分鐘後執行
eat = datetime.utcnow() + timedelta(minutes=1)
# 提交任務
res = send_sms.apply_async(args=['13855411111', '123'], eta=eta)
執行定時任務：
執行定時任務需要啟動beat和worker
- beat：定時提交任務的程序---》設定在app.conf.beat_schedule的任務
- worker：執行任務
- 第一步：在celery的py檔案中寫入
# 匯入定時需要的模組
from celery.schedules import crontab
# 第一步：在celery的py檔案中寫入
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# celery的組態檔#####
# 任務的定時設定
app.conf.beat_schedule = {
'send_sms': { # 設定執行函數的名字
'task': 'celery_task.task.send_sms', # 匯入任務的位置
# 'schedule': timedelta(seconds=3), # 時間物件
# 'schedule': crontab(hour=8, day_of_week=1), # 每週一早八點
'schedule': crontab(hour=9, minute=43), # 每天9點43
'args': ('18888888', '6666'), # 設定執行函數需要的引數
},
}
- 第二步：啟動beat # 啟動後配設定的任務會自動提交
celery -A celery_task beat -l info
- 第三步：啟動worker # beat提交的任務被被執行
celery -A celery_task worker -l info -P eventlet
三、Django中使用celery
補充：
如果在公司中，只做定時任務有一個框架更簡單一點
使用步驟：
-1 把咱們寫的包，複製到專案目錄下
-luffy_api
-celery_task #celery的包路徑
-luffy_api #原始碼路徑
-2 在使用提交非同步任務的位置，匯入使用即可
-檢視函數中使用，匯入任務
-任務.delay() # 提交任務
-3 啟動worker，如果有定時任務，啟動beat
-4 等待任務被worker執行
-5 在檢視函數中，查詢任務執行的結果
1、模擬寫一個非同步秒殺任務
後端
view.py
from celery.result import AsyncResult
from celery_task.celery import app
from celery_task.task import sckill_task
# 秒殺介面
class SeckillView(ViewSet):
# 開啟秒殺
@action(methods=['GET'], detail=False)
def seckill(self, request):
# 獲取商品連結
goods_id = request.query_params.get('goods_id')
# 將任務提交給worker
res = sckill_task.delay(goods_id)
# 將任務的ID反饋給前端
return APIResponse(task_id=str(res))
# 查詢秒殺結果
@action(methods=['GET'], detail=False)
def get_result(self, request):
# 前端將任務ID產過來，用於接收結果
task_id = request.query_params.get('task_id')
# 呼叫介面，查詢結果
a = AsyncResult(id=task_id, app=app)
if a.successful():
result = a.get()
if result:
return APIResponse(msg='秒殺成功')
else:
return APIResponse(code=101, msg='手速滿了，秒殺失敗')
elif a.status == 'PENDING':
return APIResponse(code=666, msg='加速秒殺中')
return APIResponse(msg='錯誤')
celery.py ---->秒殺任務
import random
# 秒殺函數
@app.task()
def sckill_task(goods_id):
print('商品正在秒殺中')
time.sleep(random.choice([6, 7, 8, 9]))
print('商品秒殺結束')
return random.choice([True, False])
前端：
<template>
<div>
<button @click="clickHandle">點選秒殺</button>
</div>
</template>
<script>
export default {
name: "Template",
data() {
return {
// 用於接收任務ID
task_id: '',
// 使用者存放定時任務
t: ''
}
},
methods: {
// 使用者點選秒殺後傳送請求
clickHandle() {
// 向厚點提交秒殺任務
this.$axios.get(this.$settings.BASE_URL + '/user/seckill/seckill/?goods_id=1').then(res => {
// 判斷任務是否提交成功
if (res.data.code == 100) {
// 提交成功會獲取到任務ID
this.task_id = res.data.task_id
// 告知使用者商品正在秒殺中
this.$message('正在秒殺中')
// 啟動一個定時任務，每隔3秒向後端傳送請求，獲取任務是否提交成功
this.t = setInterval(res => {
// 定時向後端傳送請求，判斷秒殺結果
this.$axios.get(this.$settings.BASE_URL + '/user/seckill/get_result/?task_id=' + this.task_id).then(res => {
// 判斷任務是否結束
if (res.data.code == 666) {
this.$message(res.data.msg)
// 任務結束反饋結果，關閉定時器
} else {
this.$message(res.data.msg)
// 關閉定時器
clearInterval(this.t)
this.t = ''
}
})
}, 3000)
}
})
}
}
}
</script>
2、總結
- 第一步：將celery包複製到專案路徑下
-luffy_api
-celery_task #celery的包路徑
celery.py # 一定不要忘了一句話
import os
# 重點：celery中使用djagno,任務中可能會使用django的orm，快取，表模型。。。。一定要加
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffy_api.settings.dev')
-luffy_api #原始碼路徑
- 第二步：在需要使用非同步的地方匯入celery範例即可使用
-檢視函數中使用，匯入任務
-任務.delay() # 提交任務
-
第三步：啟動worker，如果有定時任務，啟動beat
-
第四步： 等待任務被worker執行
-
第五步：在檢視函數中，查詢任務結果