celery介紹

celery是一個異步任務(wù)框架，它可以執(zhí)行異步任務(wù)、延遲任務(wù)、定時任務(wù)

長垣網(wǎng)站制作公司哪家好，找創(chuàng)新互聯(lián)！從網(wǎng)頁設(shè)計、網(wǎng)站建設(shè)、微信開發(fā)、APP開發(fā)、響應(yīng)式網(wǎng)站等網(wǎng)站項目制作，到程序開發(fā)，運營維護。創(chuàng)新互聯(lián)從2013年開始到現(xiàn)在10年的時間，我們擁有了豐富的建站經(jīng)驗和運維經(jīng)驗，來保證我們的工作的順利進行。專注于網(wǎng)站建設(shè)就選創(chuàng)新互聯(lián)。

異步任務(wù)框架簡述：

1）celery可以不依賴任何服務(wù)器，通過自身命令，啟動服務(wù)(內(nèi)部支持socket)
2）celery服務(wù)為為其他項目服務(wù)提供異步解決任務(wù)需求的
注：會有兩個服務(wù)同時運行，一個是項目服務(wù)，一個是celery服務(wù)，項目服務(wù)將需要異步處理的任務(wù)交給celery服務(wù)，celery就會在需要時異步完成項目的需求

Celery架構(gòu)

Celery的架構(gòu)由三部分組成，消息中間件（message broker）、任務(wù)執(zhí)行單元（worker）和 任務(wù)執(zhí)行結(jié)果存儲（task result store）組成。

消息中間件

Celery本身不提供消息服務(wù)，但是可以方便的和第三方提供的消息中間件集成。包括，RabbitMQ, Redis等等

任務(wù)執(zhí)行單元

Worker是Celery提供的任務(wù)執(zhí)行的單元，worker并發(fā)的運行在分布式的系統(tǒng)節(jié)點中。

任務(wù)結(jié)果存儲

Task result store用來存儲Worker執(zhí)行的任務(wù)的結(jié)果，Celery支持以不同方式存儲任務(wù)的結(jié)果，包括AMQP, redis等

使用場景

異步執(zhí)行：解決耗時任務(wù),將耗時操作任務(wù)提交給Celery去異步執(zhí)行，比如發(fā)送短信/郵件、消息推送、音視頻處理等等

延遲執(zhí)行：解決延遲任務(wù)

定時執(zhí)行：解決周期(周期)任務(wù),比如每天數(shù)據(jù)統(tǒng)計

Celery的安裝配置

pip install celery gevent

windows環(huán)境下啟動celery

# module為python module名
celery -A  worker -l info -P gevent

celery執(zhí)行異步任務(wù)

采用包架構(gòu)封裝（多任務(wù)結(jié)構(gòu)）

在項目文件project下新建一個celery包

project
    ├── celery_task  	# celery包
    │   ├── __init__.py # 包文件
    │   ├── celery.py   # celery連接和配置相關(guān)文件，且名字必須叫celery.py
    │   └── tasks.py    # 所有任務(wù)函數(shù)
    ├── add_task.py  	# 添加任務(wù)
    └── get_result.py   # 獲取結(jié)果

celery.py

# 導(dǎo)入模塊
from celery import Celery
# 指定任務(wù)中間件和任務(wù)處理倉庫為redis里的第一個庫，后面1表示第一個，2表示第二個。
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:@127.0.0.1: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 #添加該裝飾器表示這個函數(shù)任務(wù)使用了celery框架的異步處理
def add(n, m):
    print(n)
    print(m)
    time.sleep(10)
    print('n+m的結(jié)果：%s' % (n + m))
    return n + m

@app.task
def low(n, m):
    print(n)
    print(m)
    print('n-m的結(jié)果：%s' % (n - m))
    return n - m

add_task.py
這個py文件是用來添加任務(wù)的，可以建在任意位置

from celery_task import tasks

# 添加立即執(zhí)行任務(wù)
t1 = tasks.add.delay(10, 20)
t2 = tasks.low.delay(100, 50)
print(t1) # 獲取的就是這個任務(wù)的id號


# 添加延遲任務(wù)
from datetime import datetime, timedelta
eta=datetime.utcnow() + timedelta(seconds=10)
tasks.low.apply_async(args=(200, 50), eta=eta)

get_result.py
這個py文件是用來獲取任務(wù)處理的結(jié)果的

from celery_task.celery import app

from celery.result import AsyncResult

id = 'a40-9d32-44b5-a701-9a31cc3c74b5' #輸入任務(wù)的id號
if __name__ == '__main__':
    async = AsyncResult(id=id, app=app)
    if async.successful(): #如果這個任務(wù)執(zhí)行成功
        result = async.get() #得到結(jié)果
        print(result)
    elif async.failed():
        print('任務(wù)失敗')
    elif async.status == 'PENDING':
        print('任務(wù)等待中被執(zhí)行')
    elif async.status == 'RETRY':
        print('任務(wù)異常后正在重試')
    elif async.status == 'STARTED':
        print('任務(wù)已經(jīng)開始被執(zhí)行')

celery執(zhí)行延遲任務(wù)

在add_task.py里書寫：

# 添加延遲任務(wù)
from datetime import datetime, timedelta
# datetime.utcnow()表示獲取當(dāng)前的utc時間，timedelta(seconds=10)表示加10秒timedelta括號里之后是時間類型，可以相加減
eta=datetime.utcnow() + timedelta(seconds=10)
# 使用的是apply_async這個方法，tasks.add這個是函數(shù)名，args括號里是函數(shù)add需要傳的參數(shù)，eta必須是utc時間
tasks.add.apply_async(args=(200, 50), eta=eta)

celery執(zhí)行定時任務(wù)

將celery.py添加以下定時配置

# 導(dǎo)入模塊
from celery import Celery
# 指定任務(wù)中間件和任務(wù)處理倉庫為redis里的第一個庫，后面1表示第一個，2表示第二個。
broker = 'redis://127.0.0.1:6379/1'
backend = 'redis://127.0.0.1:6379/2'
# backend='redis://:@127.0.0.1:6379/1' redis加密碼寫法
# include列表里表示celery_task包里的task.py交給app處理
app = Celery(__name__,broker=broker, backend=backend, include=['celery_task.tasks'])


# 定時任務(wù)配置
# 時區(qū)
app.conf.timezone = 'Asia/Shanghai'
# 是否使用UTC
app.conf.enable_utc = False

# 任務(wù)的定時配置
from datetime import timedelta
from celery.schedules import crontab
app.conf.beat_schedule = {
    'add-task': {
        'task': 'celery_task.tasks.add',  # celery_task.tasks.low這個是你函數(shù)任務(wù)的位置，task是固定的
        'schedule': timedelta(seconds=3), # 每隔三秒執(zhí)行一次
        # 'schedule': crontab(hour=8, day_of_week=1),  # 每周一早八點
        'args': (300, 150), # 函數(shù)任務(wù)需要傳入的參數(shù)
    }
}

然后在終端下cd到scripts文件夾

執(zhí)行celery -A celery_task beat
表示開始定時發(fā)布任務(wù)

celery在django中的使用

需要在celery.py中添加以下代碼：

# 加載django環(huán)境
import os
import django
# luffyapi.settings.dev 這個是項目的配置文件位置
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "luffyapi.settings.dev")
django.setup()