celery筆記一之celery介紹、啟動(dòng)和運(yùn)行結(jié)果跟蹤-環(huán)球熱文

來源：博客園

本文首發(fā)于公眾號：Hunter后端原文鏈接：celery筆記一之celery介紹、啟動(dòng)和運(yùn)行結(jié)果跟蹤


【資料圖】

本篇筆記內(nèi)容如下：

celery 介紹celery 準(zhǔn)備celery 啟動(dòng)和異步任務(wù)的運(yùn)行運(yùn)行結(jié)果跟蹤1、celery 介紹

celery 大致有兩種應(yīng)用場景，一種是異步任務(wù)，一種是定時(shí)任務(wù)。

比如說在一個(gè)接口請求中，某個(gè)函數(shù)執(zhí)行所需的時(shí)間過長，而前端頁面并不是立刻需要在接口中獲取處理結(jié)果，可以將這個(gè)函數(shù)作為異步任務(wù)，先返回給前端處理中的信息，在后臺(tái)單獨(dú)運(yùn)行這個(gè)函數(shù)，這就是異步任務(wù)。

另一個(gè)比如說某個(gè)函數(shù)需要每天晚上運(yùn)行一遍，不可能人天天守著后臺(tái)手動(dòng)執(zhí)行一遍這個(gè)函數(shù)，那么就可以用 celery 來實(shí)現(xiàn)這個(gè)定時(shí)的周期任務(wù)。

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

task

這個(gè)任務(wù)就是我們前面舉的例子的異步任務(wù)或者是定時(shí)任務(wù)，即為 task，我們可以定義這些任務(wù)，然后發(fā)送到 broker

broker

broker 可以理解成消息中間件，用于獲取異步或者定時(shí)任務(wù)，形成一個(gè)或多個(gè)消息隊(duì)列，然后發(fā)送給 worker 處理這些消息

broker 的形式可以是 Redis，RabbitMQ 或者其他，這里我們使用 Redis 作為消息中間件

worker

worker 是處理消息的程序，獲取 broker 中的消息，然后在 worker 中執(zhí)行，然后根據(jù)配置決定將處理結(jié)果發(fā)送到 backend

result_backend

在 worker 處理完消息之后會(huì)有 return 或者沒有返回結(jié)果，都會(huì)根據(jù)配置將結(jié)果發(fā)送出來，可以配置成發(fā)送到 redis 中，也可以將之存儲(chǔ)到 database 中

beat

主要用于調(diào)用定時(shí)任務(wù)，根據(jù)設(shè)定好的定時(shí)任務(wù)，比如每天晚上十點(diǎn)執(zhí)行某個(gè)函數(shù)，beat 則會(huì)在相應(yīng)的時(shí)間將這個(gè) task 發(fā)送給 broker，然后 worker 獲取任務(wù)進(jìn)行處理

定時(shí)任務(wù)除了說的每天晚上十點(diǎn)這種周期任務(wù)，也可以是間隔任務(wù)，比如說每隔多少秒，多少分鐘執(zhí)行一次

注意：異步任務(wù)的發(fā)送是不經(jīng)過 beat 處理，直接發(fā)送給 broker 的

在上面的結(jié)構(gòu)中，broker 需要將相應(yīng)的服務(wù)比如 redis 運(yùn)行起來，而 worker 和 beat 需要在手動(dòng)用程序運(yùn)行，而且每次更改了定時(shí)策略之后需要重新啟動(dòng) beat 和 worker 才能生效。

2、celery 準(zhǔn)備

接下來我們實(shí)現(xiàn)一個(gè)最簡單的異步任務(wù)，在執(zhí)行異步任務(wù)前，我們做如下的準(zhǔn)備工作

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，可以去官網(wǎng)下載一個(gè) redis 啟動(dòng)，也可以使用 docker 來執(zhí)行安裝。

我在之前的 docker 系列筆記中有介紹過如何拉取鏡像和運(yùn)行容器，我們這里直接使用 docker 來運(yùn)行：

docker run -itd -p 6379:6379 redis:latest
3.異步任務(wù)準(zhǔn)備

我們準(zhǔn)備一個(gè)最簡單的 add() 函數(shù)，放在 tasks.py 文件中：

# tasks.pyfrom celery import Celeryapp = Celery("tasks", broker="redis://localhost/0", backend="redis://localhost/1")@app.taskdef add(x, y):     return x + y

在這段代碼里，我們引入 Celery 模塊，并將其實(shí)例化為 app，且配置了 broker 參數(shù)，表示消息隊(duì)列都會(huì)被放在 redis 的第一個(gè)數(shù)據(jù)庫下

指定的 backend 參數(shù)則表示函數(shù)運(yùn)行的結(jié)果被放在 redis 的第二個(gè)數(shù)據(jù)庫下

然后用 @app.task 修飾 add 函數(shù)，表示它是 app 下的 task 任務(wù)

以上，我們的準(zhǔn)備工作就完成了，接下來嘗試運(yùn)行這個(gè)異步任務(wù)

3、celery 啟動(dòng)和異步任務(wù)的運(yùn)行

說是 celery 的啟動(dòng)，其實(shí)是 worker 的啟動(dòng)，中間件是 redis，已經(jīng)在前面的步驟中啟動(dòng)了。

我們在 tasks.py 所在的文件夾下執(zhí)行下面的命令：

celery -A tasks worker -l INFO

在這里，tasks 是我們?nèi)蝿?wù)所在的文件名，worker 表示啟動(dòng)的是 worker 程序

-l INFO 則會(huì)在控制臺(tái)打印出 worker 接收到的消息詳情，如果不執(zhí)行，則信息流不會(huì)被打印出來

執(zhí)行了上面的程序后，可以看到控制臺(tái)會(huì)輸出下面這種信息：

-------------- 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 啟動(dòng)成功

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

在另一個(gè) shell 窗口，進(jìn)入 python 的交互界面，輸入以下命令：

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

add 是我們需要執(zhí)行的異步任務(wù)的函數(shù)名

delay 是異步任務(wù)執(zhí)行的特有方法，這個(gè)其實(shí)是 apply_async() 函數(shù)的簡便寫法，不帶任何參數(shù)，apply_async() 除了可以實(shí)現(xiàn)異步任務(wù)的功能，還可以指定多少時(shí)間后執(zhí)行，比如說二十秒后執(zhí)行，這個(gè)在后面的筆記中我們再介紹。

而異步任務(wù)的返回我們這里用 res 來定義，它是一個(gè)包含了這個(gè)任務(wù)所有執(zhí)行信息對象，有任務(wù)狀態(tài)（是否執(zhí)行成功），有返回結(jié)果（add() 函數(shù)的return），還有這個(gè) task 特有的標(biāo)識 id等信息

至此，我們的一個(gè)異步任務(wù)的執(zhí)行就完成了，我們可以在下一步查看它的運(yùn)行結(jié)果等信息。

4、運(yùn)行結(jié)果跟蹤

接下來，我們在 tasks.py 中建立下面幾個(gè)函數(shù)，來測試我們對結(jié)果的跟蹤：

# tasks.pyimport timefrom celery import Celeryapp = Celery("tasks", broker="redis://localhost/0", backend="redis://localhost/1")@app.taskdef add(x, y):    return x + y@app.taskdef div(x, y):    return x / y@app.taskdef test_not_finished():    time.sleep(30)    return True

然后重新運(yùn)行 worker：

celery -A tasks worker -l INFO

然后引入和執(zhí)行函數(shù)：

from tasks import add, div, test_not_finished

獲取延時(shí)任務(wù)的結(jié)果

res = add.delay(1, 2)print(res.result)# 也可以使用 get() print(res.get())

get() 函數(shù)也可以加個(gè)超時(shí)的設(shè)置：

res.get(timeout=2)

但是這樣需要注意，因?yàn)槿绻瑫r(shí)了還未獲取到結(jié)果，程序就會(huì)報(bào)錯(cuò)

判斷函數(shù)運(yùn)行是否完成

print(res.ready())

打印出的結(jié)果為 True 則表示函數(shù)運(yùn)行完成

我們可以測試函數(shù)為完成的狀態(tài)：

res2 = test_not_finished.delay()

在這個(gè)函數(shù)里，我們設(shè)置了 30s 的休眠，所以在 30s 內(nèi)我們打印結(jié)果可以看到 res2.ready() 是為 False 的：

print(res2.ready())

獲取task id

每個(gè)被執(zhí)行的 task 都有各自對應(yīng)的 id 作為它們的唯一鍵：

print(res.id)

查看任務(wù)執(zhí)行的狀態(tài)

# 任務(wù)執(zhí)行是否失敗，返回 布爾型數(shù)據(jù)is_failed = res.failed()# 任務(wù)執(zhí)行是否成功，返回布爾型數(shù)據(jù)is_successful = res.successful()# 執(zhí)行的任務(wù)所處的狀態(tài)state = res.state# state 的值會(huì)在 PENDING，STARTED，SUCCESS，RETRY，F(xiàn)AILURE 這幾種狀態(tài)中，分別是 待處理中，任務(wù)已經(jīng)開始，成功，重試中，失敗

報(bào)錯(cuò)處理

如果執(zhí)行的延時(shí)任務(wù)在程序中報(bào)錯(cuò)，比如我們定義的 div() 函數(shù)，我們傳入的除數(shù)為 0 的話，在程序中是會(huì)報(bào)錯(cuò)的，我們使用 get() 來獲取結(jié)果的話程序是會(huì)報(bào)錯(cuò)的：

res3 = div.delay(3, 0)res3.get()# 返回會(huì)報(bào)錯(cuò)

但是我們可以使用 propagate=False 參數(shù)來忽略程序的報(bào)錯(cuò):

res3.get(propagate=False)

這樣我們獲取的就不是程序報(bào)錯(cuò)，而是程序報(bào)錯(cuò)的信息作為結(jié)果返回

使用 res3.state 發(fā)現(xiàn)返回的結(jié)果是 FAILURE

當(dāng)延時(shí)任務(wù)在程序中報(bào)錯(cuò)，它的返回值就不會(huì)是正確的，我們可以通過 res3.traceback 是否有值來判斷函數(shù)運(yùn)行過程中是有報(bào)錯(cuò)：

if res3.traceback:    print("延時(shí)任務(wù)報(bào)錯(cuò)")else:    print("程序正常運(yùn)行，可以獲取返回值")

result資源釋放

因?yàn)?backend 會(huì)使用資源來保存和傳輸結(jié)果，為了確保資源被釋放，所以在執(zhí)行完異步任務(wù)后，你必須對每一個(gè)結(jié)果調(diào)用 get() 或者 forget() 函數(shù)

result.get() 函數(shù)獲取結(jié)果

result.forget() 在 backend 刪掉該數(shù)據(jù)

在官方文檔上，意思是 get() 和 forget() 方法都可以釋放資源，但是經(jīng)過我測試，貌似只有 forget() 函數(shù)會(huì)釋放資源

查看是否資源被釋放也很簡單，登錄到對應(yīng)的 backend，我這里是 redis，使用 redis-cli 或者通過 docker 進(jìn)入 redis：

select 1keys*

查看相應(yīng)的 task id 是否還在列表就可以知道該資源是否被釋放

如果不想手動(dòng)釋放資源，可以在配置里設(shè)置一個(gè)過期時(shí)間，那么結(jié)果就會(huì)在指定時(shí)間段后被釋放：

app.conf.update(result_expires=60)

這個(gè)我們可以在后面的配置里再詳細(xì)介紹。

以上就是本篇筆記全部內(nèi)容，下一篇筆記我們將介紹如何建立一個(gè) celery 項(xiàng)目、配置的幾種方法及一些基本的配置。

如果想獲取更多后端相關(guān)文章，可掃碼關(guān)注閱讀：

標(biāo)簽：

推薦

財(cái)富更多》

動(dòng)態(tài)更多》

熱點(diǎn)