目錄
- Celery介紹、安裝、基本使用
- 一、Celery服務
- 1、celery架構
- 2、celery快速使用
- 二、Celer包結構
- 1、創建clery包結構
- 2、Celery執行異步任務、延遲任務、定時任務
- 三、Django中使用celery
- 1、模擬寫一個異步秒殺任務
- 2、總結
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
補充：
如果在公司中，只做定時任務有一個框架更簡單一點
- APSchedule：https://blog.csdn.net/qq_41341757/article/details/118759836
使用步驟：
-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執行
-
第五步：在視圖函式中，查詢任務結果
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/546307.html
標籤：Python
下一篇：Python基礎_概述