celery介紹 celery是一個非同步任務框架，它可以執行非同步任務、延遲任務、定時任務 非同步任務框架簡述： 1）celery可以不依賴任何伺服器，通過自身命令，啟動服務(內部支持socket) 2）celery服務為為其他項目服務提供非同步解決任務需求的 註：會有兩個服務同時運行，一個是項目服務，一 ...
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
表示開始定時發佈任務
