Celery

Celery是一個功能完備即插即用的異步任務佇列系統，它適用于異步處理問題，當發送郵件、或者檔案上傳, 影像處理等等一些比較耗時的操作，我們可將其異步執行，這樣用戶不需要等待很久，提高用戶體驗，

檔案：http://docs.jinkan.org/docs/celery/getting-started/index.html

Celery的特點是：

• 簡單，易于使用和維護，有豐富的檔案，
• 高效，單個celery行程每分鐘可以處理數百萬個任務，
• 靈活，celery中幾乎每個部分都可以自定義擴展，

任務佇列是一種跨執行緒、跨機器作業的一種機制.
任務佇列中包含稱作任務的作業單元，有專門的作業行程持續不斷的監視任務佇列，并從中獲得新的任務并處理.
celery通過訊息(任務)進行通信，通常使用一個叫Broker(中間人)來協助clients(任務的發出者)和worker(任務的處理者). clients發出訊息到佇列中，broker將佇列中的資訊派發給worker來處理，

Celery的架構

Celery的架構由三部分組成，訊息佇列（message broker），任務執行單元（worker）和任務執行結果存盤（task result store）組成，

一個celery系統可以包含很多的worker和broker
Celery本身不提供訊息佇列功能，但是可以很方便地和第三方提供的訊息中間件進行集成，包括RabbitMQ,Redis,MongoDB等

安裝

pip install -U celery==4.4.7 -i  https://pypi.tuna.tsinghua.edu.cn/simple

注意：

Celery不建議在windows系統下使用，因為Celery在4.0版本以后，不再支持windows系統，所以如果要在windows下使用只能安裝4.0以前的版本，而且即便是4.0之前的版本，在windows系統下也是不能單獨使用的，需要安裝eventlet模塊

也可從官方直接下載安裝包:https://pypi.python.org/pypi/celery/

tar xvfz celery-4.4.7.tar.gz
cd celery-4.4.7
python setup.py build
python setup.py install

使用

使用celery第一件要做的最為重要的事情是需要先創建一個Celery實體，我們一般叫做celery應用，或者更簡單直接叫做一個app，app應用是我們使用celery所有功能的入口，比如創建任務，管理任務等，在使用celery的時候，app必須能夠被其他的模塊匯入，

一般celery任務目錄直接放在專案的根目錄下即可，路徑:

luffyapi/
├── mycelery/
    ├── config.py     # 組態檔
    ├── __init__.py   
    ├── main.py       # 主程式
    └── sms/          # 一個目錄可以放置多個任務,該目錄下存放當前任務執行時需要的模塊或依賴
        └── tasks.py  # 任務的檔案，名稱必須是這個!!!

main.py，代碼：

# 主程式
from celery import Celery
# 創建celery實體物件
app = Celery("luffy")

# 通過app物件加載配置
app.config_from_object("mycelery.config")

# 自動搜索并加載任務
# 引數必須必須是一個串列，里面的每一個任務都是任務的路徑名稱
# app.autodiscover_tasks(["任務1","任務2",....])
app.autodiscover_tasks(["mycelery.sms","mycelery.email"])

# 啟動Celery的命令
# 強烈建議切換目錄到專案的根目錄下啟動celery!!
# celery -A mycelery.main worker --loglevel=info

組態檔config.py，代碼：

# 任務佇列的鏈接地址
broker_url = 'redis://127.0.0.1:6379/15'
# 結果佇列的鏈接地址
result_backend = 'redis://127.0.0.1:6379/14'

創建一個任務檔案sms/tasks.py，并創建任務，代碼:

# celery的任務必須寫在tasks.py的檔案中，別的檔案名稱不識別!!!
from mycelery.main import app

@app.task  # name表示設定任務的名稱，如果不填寫，則默認使用函式名做為任務名
def send_sms():
    print("發送短信!!!")

@app.task(name="send_sms2")  # name表示設定任務的名稱，如果不填寫，則默認使用函式名做為任務名
def send_sms2():
    print("發送短信任務2!!!")

接下來，我們運行celery，效果如下：

在程式中呼叫上面的異步任務，拿django進行舉例：

# 呼叫celery執行異步任務
from my_celery.sms.tasks import send_sms
send_sms.delay(mobile)

其他參考檔案：

http://docs.celeryproject.org/en/latest/getting-started/introduction.html

https://github.com/celery/celery/tree/master/examples/django/

https://www.jianshu.com/p/1840035cb510

https://flower.readthedocs.io/en/latest/screenshots.html

接下來，我們需要把celery和django組合起來一起使用，

把django和celery進行組合

在main.py主程式中對django的組態檔進行加載

import os,django
from celery import Celery
# 初始化celery物件
app = Celery("luffy")

# 初始化django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'luffyapi.settings.dev')
django.setup()

# 加載配置
app.config_from_object("mycelery.config")
# 自動注冊任務
app.autodiscover_tasks(["mycelery.sms","mycelery.email"])
# 運行celery
# 終端下: celery -A mycelery.main worker -l info

在需要使用django配置的任務中，直接加載配置，所以我們把注冊的短信發送功能，整合成一個任務函式，mycelery.sms.tasks，代碼：

import json,logging
from mycelery.main import app
from ronglian_sms_sdk import SmsSDK
from django.conf import settings
from luffyapi.settings import constants

log = logging.getLogger("django")

@app.task(name="send_sms")
def send_sms(mobile,code):
    """發送短信"""
    sdk = SmsSDK(
        settings.SMS.get("accId"),
        settings.SMS.get("accToken"),
        settings.SMS.get("appId"),
    )

    try:
        resp = sdk.sendMessage(settings.SMS.get("TempId"), mobile, (code, constants.SMS_EXPIRE_TIME // 60))
        resp_data = https://www.cnblogs.com/wylshkjj/archive/2020/10/15/json.loads(resp)
        if resp_data.get("statusCode") != "000000":
            raise Exception
    except Exception as exc:
        log.error("短信發送失敗! 手機號:%s: %s" % (mobile,exc) )
        raise Exception

在這個任務中，我們需要加載短信發送的sdk和相關的配置常量，所以我們可以直接把django中的短信發送模塊和相關的常量組態檔直接剪切到當前sms任務目錄中

mycelery/
├── config.py
├── __init__.py
├── main.py
└── sms/
    ├── __init__.py
    ├── tasks.py

再次啟動專案即可，

最終在django里面，我們呼叫Celery來異步執行任務，需要完成2個步驟：

# 1. 宣告一個和celery一模一樣的任務函式，但是我們可以導包來解決
from mycelery.sms.tasks import send_sms

# 2. 呼叫任務函式，發布任務
send_sms.delay(mobile,code)
# send_sms.delay() 如果呼叫的任務函式沒有引數，則不需要填寫任何內容

標籤：其他

上一篇：零基礎想學好編程！C語言最難啃的 3 塊硬骨頭，你全吃透了嗎？

下一篇：Mock測驗你的Spring MVC介面