celery筆記一之celery介紹、啟動和執行結果跟蹤

本文首發於公眾號：Hunter後端
原文連結：celery筆記一之celery介紹、啟動和執行結果跟蹤

本篇筆記內容如下：

1. celery 介紹
2. celery 準備
3. celery 啟動和非同步任務的執行
4. 執行結果跟蹤

1、celery 介紹

celery 大致有兩種應用場景，一種是非同步任務，一種是定時任務。

比如說在一個介面請求中，某個函數執行所需的時間過長，而前端頁面並不是立刻需要在介面中獲取處理結果，可以將這個函數作為非同步任務，先返回給前端處理中的資訊，在後臺單獨執行這個函數，這就是非同步任務。

另一個比如說某個函數需要每天晚上執行一遍，不可能人天天守著後臺手動執行一遍這個函數，那麼就可以用 celery 來實現這個定時的週期任務。

接下來介紹一下 celery 的組成：

task

這個任務就是我們前面舉的例子的非同步任務或者是定時任務，即為 task，我們可以定義這些任務，然後傳送到 broker

broker

broker 可以理解成訊息中介軟體，用於獲取非同步或者定時任務，形成一個或多個訊息佇列，然後傳送給 worker 處理這些訊息

broker 的形式可以是 Redis，RabbitMQ 或者其他，這裡我們使用 Redis 作為訊息中介軟體

worker

worker 是處理訊息的程式，獲取 broker 中的訊息，然後在 worker 中執行，然後根據設定決定將處理結果傳送到 backend

result_backend

在 worker 處理完訊息之後會有 return 或者沒有返回結果，都會根據設定將結果傳送出來，可以設定成傳送到 redis 中，也可以將之儲存到 database 中

beat

主要用於呼叫定時任務，根據設定好的定時任務，比如每天晚上十點執行某個函數，beat 則會在相應的時間將這個 task 傳送給 broker，然後 worker 獲取任務進行處理

定時任務除了說的每天晚上十點這種週期任務，也可以是間隔任務，比如說每隔多少秒，多少分鐘執行一次

注意：非同步任務的傳送是不經過 beat 處理，直接傳送給 broker 的

在上面的結構中，broker 需要將相應的服務比如 redis 執行起來，而 worker 和 beat 需要在手動用程式執行，而且每次更改了定時策略之後需要重新啟動 beat 和 worker 才能生效。

2、celery 準備

接下來我們實現一個最簡單的非同步任務，在執行非同步任務前，我們做如下的準備工作

1.安裝依賴

我們需要安裝一下 celery 和 redis 的依賴：

pip3 install celery==5.1.2 -i https://mirrors.aliyun.com/pypi/simple/
pip3 install redis==3.5.3 -i https://mirrors.aliyun.com/pypi/simple/

2.訊息中介軟體

這裡我們用到的訊息中介軟體是 redis，可以去官網下載一個 redis 啟動，也可以使用 docker 來執行安裝。

我在之前的 docker 系列筆記中有介紹過如何拉取映象和執行容器，我們這裡直接使用 docker 來執行：

docker run -itd -p 6379:6379 redis:latest

3.非同步任務準備

我們準備一個最簡單的 add() 函數，放在 tasks.py 檔案中：

# tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/1')

@app.task
def add(x, y): 
    return x + y

在這段程式碼裡，我們引入 Celery 模組，並將其範例化為 app，且設定了 broker 引數，表示訊息佇列都會被放在 redis 的第一個資料庫下

指定的 backend 引數則表示函數執行的結果被放在 redis 的第二個資料庫下

然後用 @app.task 修飾 add 函數，表示它是 app 下的 task 任務

以上，我們的準備工作就完成了，接下來嘗試執行這個非同步任務

3、celery 啟動和非同步任務的執行

說是 celery 的啟動，其實是 worker 的啟動，中介軟體是 redis，已經在前面的步驟中啟動了。

我們在 tasks.py 所在的資料夾下執行下面的命令：

celery -A tasks worker -l INFO

在這裡，tasks 是我們任務所在的檔名，worker 表示啟動的是 worker 程式

-l INFO 則會在控制檯列印出 worker 接收到的訊息詳情，如果不執行，則資訊流不會被列印出來

執行了上面的程式後，可以看到控制檯會輸出下面這種資訊：

 -------------- celery@localhost v5.1.2 (sun-harmonics)
--- ***** ----- 
-- ******* ---- Darwin-21.4.0-x86_64-i386-64bit 2022-07-17 23:56:09
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7fc8ddf3df98
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . tasks.add

[2022-07-17 23:56:09,685: INFO/MainProcess] Connected to redis://localhost:6379/0
[2022-07-17 23:56:09,699: INFO/MainProcess] mingle: searching for neighbors
[2022-07-17 23:56:10,737: INFO/MainProcess] mingle: all alone
[2022-07-17 23:56:10,780: INFO/MainProcess] celery@localhost ready.

則表示 worker 啟動成功

執行非同步任務

在另一個 shell 視窗，進入 python 的互動介面，輸入以下命令：

from tasks import add
res = add.delay(1,2)

add 是我們需要執行的非同步任務的函數名

delay 是非同步任務執行的特有方法，這個其實是 apply_async() 函數的簡便寫法，不帶任何引數，apply_async() 除了可以實現非同步任務的功能，還可以指定多少時間後執行，比如說二十秒後執行，這個在後面的筆記中我們再介紹。

而非同步任務的返回我們這裡用 res 來定義，它是一個包含了這個任務所有執行資訊物件，有任務狀態（是否執行成功），有返回結果（add() 函數的return），還有這個 task 特有的標識 id等資訊

至此，我們的一個非同步任務的執行就完成了，我們可以在下一步檢視它的執行結果等資訊。

4、執行結果跟蹤

接下來，我們在 tasks.py 中建立下面幾個函數，來測試我們對結果的跟蹤：

# tasks.py

import time
from celery import Celery

app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/1')


@app.task
def add(x, y):
    return x + y


@app.task
def div(x, y):
    return x / y


@app.task
def test_not_finished():
    time.sleep(30)
    return True

然後重新執行 worker：

celery -A tasks worker -l INFO

然後引入和執行函數：

from tasks import add, div, test_not_finished

獲取延時任務的結果

res = add.delay(1, 2)
print(res.result)

# 也可以使用 get() 
print(res.get())

get() 函數也可以加個超時的設定：

res.get(timeout=2)

但是這樣需要注意，因為如果超時了還未獲取到結果，程式就會報錯

判斷函數執行是否完成

print(res.ready())

列印出的結果為 True 則表示函數執行完成

我們可以測試函數為完成的狀態：

res2 = test_not_finished.delay()

在這個函數裡，我們設定了 30s 的休眠，所以在 30s 內我們列印結果可以看到 res2.ready() 是為 False 的：

print(res2.ready())

獲取task id

每個被執行的 task 都有各自對應的 id 作為它們的唯一鍵：

print(res.id)

檢視任務執行的狀態

# 任務執行是否失敗，返回 布林型資料
is_failed = res.failed()

# 任務執行是否成功，返回布林型資料
is_successful = res.successful()

# 執行的任務所處的狀態
state = res.state
# state 的值會在 PENDING，STARTED，SUCCESS，RETRY，FAILURE 這幾種狀態中，分別是 待處理中，任務已經開始，成功，重試中，失敗

報錯處理

如果執行的延時任務在程式中報錯，比如我們定義的 div() 函數，我們傳入的除數為 0 的話，在程式中是會報錯的，我們使用 get() 來獲取結果的話程式是會報錯的：

res3 = div.delay(3, 0)
res3.get()

# 返回會報錯

但是我們可以使用 propagate=False 引數來忽略程式的報錯:

res3.get(propagate=False)

這樣我們獲取的就不是程式報錯，而是程式報錯的資訊作為結果返回

使用 res3.state 發現返回的結果是 FAILURE

當延時任務在程式中報錯，它的返回值就不會是正確的，我們可以通過 res3.traceback 是否有值來判斷函數執行過程中是有報錯：

if res3.traceback:
    print("延時任務報錯")
else:
    print("程式正常執行，可以獲取返回值")

result資源釋放

因為 backend 會使用資源來儲存和傳輸結果，為了確保資源被釋放，所以在執行完非同步任務後，你必須對每一個結果呼叫 get() 或者 forget() 函數

result.get() 函數獲取結果

result.forget() 在 backend 刪掉該資料

在官方檔案上，意思是 get() 和 forget() 方法都可以釋放資源，但是經過我測試，貌似只有 forget() 函數會釋放資源

檢視是否資源被釋放也很簡單，登入到對應的 backend，我這裡是 redis，使用 redis-cli 或者通過 docker 進入 redis：

select 1

keys*

檢視相應的 task id 是否還在列表就可以知道該資源是否被釋放

如果不想手動釋放資源，可以在設定裡設定一個過期時間，那麼結果就會在指定時間段後被釋放：

app.conf.update(result_expires=60)

這個我們可以在後面的設定裡再詳細介紹。

以上就是本篇筆記全部內容，下一篇筆記我們將介紹如何建立一個 celery 專案、設定的幾種方法及一些基本的設定。

如果想獲取更多後端相關文章，可掃碼關注閱讀：