celery介紹
celery是一個異步任務框架，它可以執行異步任務、延遲任務、定時任務
異步任務框架簡述：
1）celery可以不依賴任何服務器，通過自身命令，啟動服務(內部支持socket)
2）celery服務為為其他專案服務提供異步解決任務需求的
注：會有兩個服務同時運行，一個是專案服務，一個是celery服務，專案服務將需要異步處理的任務交給celery服務，celery就會在需要時異步完成專案的需求
Celery架構
Celery的架構由三部分組成，訊息中間件（message broker）、任務執行單元（worker）和 任務執行結果存盤（task result store）組成，
訊息中間件
Celery本身不提供訊息服務，但是可以方便的和第三方提供的訊息中間件集成，包括，RabbitMQ, Redis等等
任務執行單元
Worker是Celery提供的任務執行的單元，worker并發的運行在分布式的系統節點中，
任務結果存盤
Task result store用來存盤Worker執行的任務的結果，Celery支持以不同方式存盤任務的結果，包括AMQP, redis等
使用場景
異步執行：解決耗時任務,將耗時操作任務提交給Celery去異步執行，比如發送短信/郵件、訊息推送、音視頻處理等等
延遲執行：解決延遲任務
定時執行：解決周期(周期)任務,比如每天資料統計
Celery的安裝配置
pip install celery gevent
windows環境下啟動celery
# module為python module名
celery -A <module> worker -l info -P gevent
celery執行異步任務
采用包架構封裝（多任務結構）
在專案檔案project下新建一個celery包
project
├── celery_task # celery包
│ ├── __init__.py # 包檔案
│ ├── celery.py # celery連接和配置相關檔案，且名字必須叫celery.py
│ └── tasks.py # 所有任務函式
├── add_task.py # 添加任務
└── get_result.py # 獲取結果
celery.py
# 匯入模塊
from celery import Celery
# 指定任務中間件和任務處理倉庫為redis里的第一個庫，后面1表示第一個，2表示第二個，
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:[email protected]:6379/1' redis加密碼寫法
# include串列里表示celery_task包里的task.py交給app處理
app = Celery(__name__,broker=broker, backend=backend, include=['celery_task.tasks'])
task.py
from .celery import app
import time
@app.task #添加該裝飾器表示這個函式任務使用了celery框架的異步處理
def add(n, m):
print(n)
print(m)
time.sleep(10)
print('n+m的結果：%s' % (n + m))
return n + m
@app.task
def low(n, m):
print(n)
print(m)
print('n-m的結果：%s' % (n - m))
return n - m
add_task.py
這個py檔案是用來添加任務的，可以建在任意位置
from celery_task import tasks
# 添加立即執行任務
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1) # 獲取的就是這個任務的id號
# 添加延遲任務
from datetime import datetime, timedelta
eta=datetime.utcnow() + timedelta(seconds=10)
tasks.low.apply_async(args=(200, 50), eta=eta)
get_result.py
這個py檔案是用來獲取任務處理的結果的
from celery_task.celery import app
from celery.result import AsyncResult
id = '21325a40-9d32-44b5-a701-9a31cc3c74b5' #輸入任務的id號
if __name__ == '__main__':
async = AsyncResult(id=id, app=app)
if async.successful(): #如果這個任務執行成功
result = async.get() #得到結果
print(result)
elif async.failed():
print('任務失敗')
elif async.status == 'PENDING':
print('任務等待中被執行')
elif async.status == 'RETRY':
print('任務例外后正在重試')
elif async.status == 'STARTED':
print('任務已經開始被執行')
celery執行延遲任務
在add_task.py里書寫：
# 添加延遲任務
from datetime import datetime, timedelta
# datetime.utcnow()表示獲取當前的utc時間，timedelta(seconds=10)表示加10秒timedelta括號里之后是時間型別，可以相加減
eta=datetime.utcnow() + timedelta(seconds=10)
# 使用的是apply_async這個方法，tasks.add這個是函式名，args括號里是函式add需要傳的引數，eta必須是utc時間
tasks.add.apply_async(args=(200, 50), eta=eta)
celery執行定時任務
將celery.py添加以下定時配置
# 匯入模塊
from celery import Celery
# 指定任務中間件和任務處理倉庫為redis里的第一個庫，后面1表示第一個，2表示第二個，
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:[email protected]:6379/1' redis加密碼寫法
# include串列里表示celery_task包里的task.py交給app處理
app = Celery(__name__,broker=broker, backend=backend, include=['celery_task.tasks'])
# 定時任務配置
# 時區
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False
# 任務的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
'add-task': {
'task': 'celery_task.tasks.add', # celery_task.tasks.low這個是你函式任務的位置，task是固定的
'schedule': timedelta(seconds=3), # 每隔三秒執行一次
# 'schedule': crontab(hour=8, day_of_week=1), # 每周一早八點
'args': (300, 150), # 函式任務需要傳入的引數
}
}
然后在終端下cd到scripts檔案夾
執行celery -A celery_task beat
表示開始定時發布任務
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/527788.html
標籤：Python
上一篇：重溫Python基礎——變數