[[439191]]

FastAPI 是 Python Web 領(lǐng)域非常受歡迎的框架，目前 GitHub 上有 39.1k 的 star，已經(jīng)遠(yuǎn)超過了 Django rest framework(22.3k star)。而 Celey 又是異步任務(wù)最流行的框架，常用于數(shù)據(jù)挖掘和機(jī)器學(xué)習(xí)等計(jì)算密集型任務(wù)的場(chǎng)景中。如果需要通過 API 來異步調(diào)用任務(wù)，那這兩個(gè)框架可以放在一起工作。本文來分享一下如何讓 FastAPI 和 Celery 更好的相互配合，開發(fā)環(huán)境下如何通過一個(gè)命令就可以讓兩者一起工作。

0、安裝依賴

pip install fastapi celery uvicorn 

1、寫個(gè)純 celery 任務(wù)

首先，讓我們來寫一個(gè)純屬 celery 的任務(wù)，讓它正常運(yùn)行，然后在通過 fastapi 來調(diào)用它。

假設(shè)你的機(jī)器已經(jīng)安裝了 Python3 和 celery，并且本機(jī)已經(jīng)開啟了 redis，運(yùn)行在 6379 端口上。

現(xiàn)在讓我們寫一個(gè)簡(jiǎn)單的任務(wù)：計(jì)算兩數(shù)之和，文件名為：celery_app.py 代碼如下：

#!/Users/aaron/py38env/bin/python 
# filename: celery_app.py 
from celery import Celery 
 
app = Celery("tasks", broker='redis://127.0.0.1:6379/0', backend='redis://127.0.0.1:6379/0') 
 
@app.task 
def add(x, y): 
    return x + y 

然后啟動(dòng)一個(gè) worker 接收遠(yuǎn)程調(diào)用。

celery -A celery_app worker -l info 

如果要遠(yuǎn)程異步調(diào)用這個(gè) add 函數(shù)，我們需要再編寫一個(gè)文件 start_celery_app.py，內(nèi)容如下：

from celery_app import add #導(dǎo)入我們的任務(wù)函數(shù)add 
import time 
result = add.delay(12,12) #異步調(diào)用，這一步不會(huì)阻塞，程序會(huì)立即往下運(yùn)行 
 
while not result.ready():# 循環(huán)檢查任務(wù)是否執(zhí)行完畢 
    print(time.strftime("%H:%M:%S")) 
    time.sleep(1) 
 
print(result.get()) #獲取任務(wù)的返回結(jié)果 
print(result.successful()) #判斷任務(wù)是否成功執(zhí)行 

任務(wù)返回了結(jié)果 24，命令成功完成，

此時(shí) worker 界面增加的信息如下：

2、通過 fastapi 來執(zhí)行

編寫一個(gè) api.py 通過接口來調(diào)用上述的 add 函數(shù)：

from fastapi import FastAPI 
import celery_app 
 
app = FastAPI() 
 
@app.get("/") 
def read_root(): 
    result = celery_app.add.delay(12, 12) 
    return {"12+12": result.get()} 

啟動(dòng)服務(wù)：

uvicorn api:app --host 0.0.0.0 --port 8000 --reload 

然后訪問：http://127.0.0.1:8000，會(huì)發(fā)現(xiàn)任務(wù)成功執(zhí)行：

這樣我們啟動(dòng)了兩個(gè)獨(dú)立的進(jìn)程，一個(gè)是 celery 的 worker，一個(gè)是 fastapi 的 app，這樣做沒問題，且生產(chǎn)環(huán)境下是嚴(yán)格要求分開運(yùn)行的，不過，開發(fā)環(huán)境下如果這樣做就太低效了。

3、開發(fā)環(huán)境下如何一條命令啟動(dòng)

如果不使用兩個(gè)終端來啟動(dòng)兩個(gè)命令，我們可以使用 Celery 提供的測(cè)試實(shí)用程序在后臺(tái)線程中啟動(dòng) celery worker，比如寫一個(gè)這樣的文件run.py，內(nèi)容如下：

import uvicorn 
 
original_callback = uvicorn.main.callback 
 
def callback(**kwargs): 
    from celery.contrib.testing.worker import start_worker 
    from celery_app import app 
 
    with start_worker(app, perform_ping_check=False, loglevel="info"): 
        original_callback(**kwargs) 
 
uvicorn.main.callback = callback 
 
 
if __name__ == "__main__": 
    uvicorn.main() 

這樣，只需要執(zhí)行一條命令就可以同時(shí)啟動(dòng) celery worker 和 fastapi 接口服務(wù)，調(diào)試的時(shí)候是不是非常方便：圖片

最后的話

本文分享了 fastapi 和 celery 是如何配合工作的，并分享了一個(gè)用于開發(fā)環(huán)境的腳本，可以通過一個(gè)命令來啟動(dòng) celery worker 和 fastapi，可能不是完美的解決方案，但確實(shí)提升了開發(fā)效率，我覺得這是值得的，如果有幫助還請(qǐng)點(diǎn)贊、在看、關(guān)注，感謝閱讀。

本文轉(zhuǎn)載自微信公眾號(hào)「Python七號(hào)」，可以通過以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系Python七號(hào)公眾號(hào)。