Django 2.1.7 集成Celery 4.3.0 從介紹到入門

原文連接：Django 2.1.7 集成Celery 4.3.0 從介紹到入門

相關篇章：Django 2.1.7 Celery 4.3.0 示例，解決Task handler raised error: ValueError('not enough values to unp...

問題拋出

咱們在作網站後端程序開發時，會碰到這樣的需求：用戶須要在咱們的網站填寫註冊信息，咱們發給用戶一封註冊激活郵件到用戶郵箱，若是因爲各類緣由，這封郵件發送所需時間較長，那麼客戶端將會等待好久，形成很差的用戶體驗.

那麼怎麼解決這樣的問題呢?

咱們將耗時任務放到後臺異步執行。不會影響用戶其餘操做。除了註冊功能，例如上傳，圖形處理等等耗時的任務，均可以按照這種思路來解決。 如何實現異步執行任務呢？咱們可以使用celery. celery除了剛纔所涉及到的異步執行任務以外，還能夠實現定時處理某些任務。

celery介紹

Celery是一個功能完備即插即用的任務隊列。它使得咱們不須要考慮複雜的問題，使用很是簡單。celery看起來彷佛很龐大，咱們先對其進行簡單的瞭解，而後再去學習其餘一些高級特性。 celery適用異步處理問題，當發送郵件、或者文件上傳, 圖像處理等等一些比較耗時的操做，咱們可將其異步執行，這樣用戶不須要等待好久，提升用戶體驗。 celery的特色是：

• 簡單，易於使用和維護，有豐富的文檔。
• 高效，單個celery進程每分鐘能夠處理數百萬個任務。
• 靈活，celery中幾乎每一個部分均可以自定義擴展。

celery很是易於集成到一些web開發框架中.

下面來看看一些概念。

Task Queue 任務隊列

任務隊列是一種跨線程、跨機器工做的一種機制.

任務隊列中包含稱做任務的工做單元。有專門的工做進程持續不斷的監視任務隊列，並從中得到新的任務並處理.

celery經過消息進行通訊，一般使用一個叫Broker(中間人)來協client(任務的發出者)和worker(任務的處理者). clients發出消息到隊列中，broker將隊列中的信息派發給worker來處理。

一個celery系統能夠包含不少的worker和broker，可加強橫向擴展性和高可用性能。

Celery安裝

咱們可使用python的包管理器pip來安裝:

pip install -U Celery
複製代碼

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

tar xvfz celery-0.0.0.tar.gz
cd celery-0.0.0
python setup.py build
python setup.py install
複製代碼

除了安裝Celery，由於本次示例須要在django 2.1.7中使用。因此還要安裝其餘庫。

全部庫安裝命令：

# 安裝最新版本celery
pip3 install -U Celery

# 安裝django集成celery
pip3 install django-celery==3.3.0

# 安裝celery的相關依賴庫，用於鏈接redis、rabbitmq等等
pip3 install "celery[librabbitmq,redis,auth,msgpack]"

# 解決win10啓動worker問題
pip3 install eventlet
複製代碼

版本信息

• python 3.7.2 或者 3.7.1 （當前我使用這兩個版本測試）
• celery 4.3.0
• django-celery 3.3.0

Borker 中間人代理

Celery須要一種解決消息的發送和接受的方式，咱們把這種用來存儲消息的的中間裝置叫作message broker, 也可叫作消息中間人。 做爲中間人，咱們有幾種方案可選擇：

1.RabbitMQ

RabbitMQ是一個功能完備，穩定的而且易於安裝的broker. 它是生產環境中最優的選擇。使用RabbitMQ的細節參照如下連接： docs.celeryproject.org/en/latest/g…

若是咱們使用的是Ubuntu或者Debian發行版的Linux，能夠直接經過下面的命令安裝RabbitMQ: sudo apt-get install rabbitmq-server 安裝完畢以後，RabbitMQ-server服務器就已經在後臺運行。若是您用的並非Ubuntu或Debian, 能夠在如下網址： www.rabbitmq.com/download.ht… 去查找本身所須要的版本軟件。

2.Redis

Redis也是一款功能完備的broker可選項，可是其更可能因意外中斷或者電源故障致使數據丟失的狀況。 關因而有那個Redis做爲Broker，可訪下面網址： docs.celeryproject.org/en/latest/g…

Application 應用

使用celery第一件要作的最爲重要的事情是須要先建立一個Celery實例，咱們通常叫作celery應用，或者更簡單直接叫作一個app。app應用是咱們使用celery全部功能的入口，好比建立任務，管理任務等，在使用celery的時候，app必須可以被其餘的模塊導入。

1.建立應用

首先在Django項目中建立一個celery_tasks文件夾，再建立tasks.py模塊, 以下：

編寫tasks.py 其內容爲:

from celery import Celery

# 使用redis做爲broker
app = Celery('celery_tasks.tasks', broker='redis://127.0.0.1:6379/8')

# 建立任務函數
@app.task
def my_task():
    print("任務函數正在執行....")
複製代碼

Celery第一個參數是給其設定一個名字， 第二參數咱們設定一箇中間人broker, 在這裏咱們使用Redis做爲中間人。my_task函數是咱們編寫的一個任務函數， 經過加上裝飾器app.task, 將其註冊到broker的隊列中。

如今咱們在建立一個worker， 等待處理隊列中的任務.打開終端，cd到tasks.py同級目錄中，執行命令: celery -A celery_tasks.tasks worker -l info -P eventlet

啓動過程以下：

(venv) F:\pythonProject\django-pratice>celery -A celery_tasks.tasks worker -l info -P eventlet

 -------------- celery@USC2VG2F9NPB650 v4.3.0 (rhubarb)
---- **** -----
--- * ***  * -- Windows-10-10.0.17763-SP0 2019-08-03 00:33:17
-- * - **** ---
- ** ---------- [config]
- ** ---------- .> app:         celery_tasks.tasks:0x278a66264a8
- ** ---------- .> transport:   redis://127.0.0.1:6379/8
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (eventlet)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery


[tasks]
  . celery_tasks.tasks.my_task

[2019-08-03 00:33:17,385: INFO/MainProcess] Connected to redis://127.0.0.1:6379/8
[2019-08-03 00:33:17,425: INFO/MainProcess] mingle: searching for neighbors
[2019-08-03 00:33:18,665: INFO/MainProcess] mingle: all alone
[2019-08-03 00:33:18,744: INFO/MainProcess] pidbox: Connected to redis://127.0.0.1:6379/8.
[2019-08-03 00:33:18,805: INFO/MainProcess] celery@USC2VG2F9NPB650 ready.
複製代碼

2.調用任務

任務加入到broker隊列中，以便剛纔咱們建立的celery workder服務器可以從隊列中取出任務並執行。如何將任務函數加入到隊列中，可以使用delay()。

進入python終端, 執行以下代碼:

In [3]: from celery_tasks.tasks import my_task

# 調用一個任務函數，將會返回一個AsyncResult對象，這個對象能夠用來檢查任務的狀態或者得到任務的返回值。
In [4]: my_task.delay()
Out[4]: <AsyncResult: 647b2589-95d2-45c9-a9a7-0b5530caf249>
複製代碼

返回worker的終端界面，查看任務執行狀況，以下：

能夠看到已經收到任務，並執行打印了信息。

3.存儲結果

若是咱們想跟蹤任務的狀態，Celery須要將結果保存到某個地方。有幾種保存的方案可選:SQLAlchemy、Django ORM、Memcached、 Redis、RPC (RabbitMQ/AMQP)。

例子咱們仍然使用Redis做爲存儲結果的方案，任務結果存儲配置咱們經過Celery的backend參數來設定。咱們將tasks模塊修改以下:

from celery import Celery

app = Celery('celery_tasks.tasks',
             broker='redis://127.0.0.1:6379/8',
             backend='redis://127.0.0.1:6379/9',)

@app.task
def my_task(a, b):
    print("任務函數正在執行....")
    return a + b
複製代碼

我給Celery增長了backend參數，指定redis做爲結果存儲,並將任務函數修改成兩個參數，而且有返回值。

下面再來執行調用一下這個任務看看。

In [1]: from celery_tasks.tasks import my_task

# 傳遞參數至任務中
In [5]: ret = my_task.delay(10,20)

# 查詢返回值的結果
In [6]: ret.result
Out[6]: 30

# 查看是否執行失敗
In [7]: ret.failed()
Out[7]: False

複製代碼

再來看看worker的執行狀況，以下：

能夠看到celery任務已經執行成功了。

要注意的是若是修改了任務代碼，那麼就須要重啓celery的worker從新註冊，否則不會生效。

更多關於result對象信息，請參閱下列網址:docs.celeryproject.org/en/latest/r…