今天就跟大家聊聊有關(guān)redis中如何配置Celery，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

創(chuàng)新互聯(lián)建站服務(wù)項目包括西雙版納網(wǎng)站建設(shè)、西雙版納網(wǎng)站制作、西雙版納網(wǎng)頁制作以及西雙版納網(wǎng)絡(luò)營銷策劃等。多年來，我們專注于互聯(lián)網(wǎng)行業(yè)，利用自身積累的技術(shù)優(yōu)勢、行業(yè)經(jīng)驗、深度合作伙伴關(guān)系等，向廣大中小型企業(yè)、政府機構(gòu)等提供互聯(lián)網(wǎng)行業(yè)的解決方案，西雙版納網(wǎng)站推廣取得了明顯的社會效益與經(jīng)濟效益。目前，我們服務(wù)的客戶以成都為中心已經(jīng)輻射到西雙版納省份的部分城市，未來相信會繼續(xù)擴大服務(wù)區(qū)域并繼續(xù)獲得客戶的支持與信任！

以下撇開Web框架，介紹基于Redis配置Celery任務(wù)的方法。

pip install celery[redis]

項目結(jié)構(gòu)

$ tree your_project  your_project  ├── __init__.py  ├── main.py  ├── celery.py  └── tasks.py  0 directories, 4 files

其中，main.py是觸發(fā)Task的業(yè)務(wù)代碼。當然，文件名可以隨意改。celery.py是Celery的app定義的位置，tasks.py是Task定義的位置，文件名不建議修改。

配置Celery

在celery.py中寫入如下代碼：

from celery import Celery  from .settings import REDIS_URL  APP = Celery(      main=__package__,      broker=REDIS_URL,      backend=REDIS_URL,      include=[f'{__package__}.tasks'],  )  APP.conf.update(task_track_started=True)

其中，REDIS_URL從同一的配置settings.py中引入， 形式大概是redis://localhost:6379/0。這里既用Redis來當broker，又用來當backend。即，既當消息隊列，又當結(jié)果反饋的數(shù)據(jù)庫（默認僅保存1天）。

在include=，需要填一個下游worker的包名列表。這里選擇了同一個包的tasks.py文件。

額外設(shè)置的task_track_started，是命令Worker反饋STARTED狀態(tài)。默認情況下，是無法知道任務(wù)什么時候開始執(zhí)行的。

編寫任務(wù)并調(diào)用

在tasks.py文件中，添加異步任務(wù)的實現(xiàn)。

from .celery import APP  @APP.task  def do_sth():      pass

在需要發(fā)起任務(wù)的地方，用.apply_async可以觸發(fā)異步調(diào)用。即，實際只是向消息隊列發(fā)送消息，真正的執(zhí)行操作在遠程。

from celery.result import AsyncResult  from .tasks imprt do_sth  result = do_sth.apply_async()  assert isinstance(result, AsyncResult)

運行Worker：

celery -A your_project worker

運行原理

一次Task從觸發(fā)到完成，序列圖如下：

其中，main代表業(yè)務(wù)代碼主進程。它可能是Django、Flask這類Web服務(wù)，也可能是一個其它類型的進程。worker就是指Celery的Worker。

main發(fā)送消息后，會得到一個AsyncResult，其中包含task_id。僅通過task_id，也可以自己構(gòu)造一個AsyncResult，查詢相關(guān)信息。其中，代表運行過程的，主要是state。

worker會持續(xù)保持對Redis（或其它消息隊列，如RabbitMQ）的關(guān)注，查詢新的消息。如果獲得新消息，將其消費后，開始運行do_sth。運行完成會把返回值對應(yīng)的結(jié)果，以及一些運行信息，回寫到Redis（或其它backend，如Django數(shù)據(jù)庫等）上。在系統(tǒng)的任何地方，通過對應(yīng)的AsyncResult(task_id)就可以查詢到結(jié)果。

Celery Task的狀態(tài)

以下是狀態(tài)圖：

其中，除SUCCESS外，還有失?。‵AILURE）、取消（REVOKED）兩個結(jié)束狀態(tài)。而RETRY則是在設(shè)置了重試機制后，進入的臨時等待狀態(tài)。

另外，如果保存在Redis的結(jié)果信息被清理（默認僅保存1天），那么任務(wù)狀態(tài)又會變成PENDING。這在設(shè)計上是個巨大的問題，使用時要做對應(yīng)容錯。

常見控制操作

result = AsyncResult(task_id)  # 阻塞等待返回  result.wait()  # 取消任務(wù)  result.revoke()  # 刪除任務(wù)記錄  result.forget()

有時，在業(yè)務(wù)主進程中需要等待異步運行的結(jié)果，這時需要使用wait。如果要取消一個排隊中、或已執(zhí)行的任務(wù)，則可以使用revoke。即使任務(wù)已經(jīng)執(zhí)行完成，也可以使用revoke，但不會有任何變化。如果需要提前刪除任務(wù)記錄，可以使用forget。

看完上述內(nèi)容，你們對Redis中如何配置Celery有進一步的了解嗎？如果還想了解更多知識或者相關(guān)內(nèi)容，請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝大家的支持。