Flask 是 Python 中有名的輕量級同步 web 框架,在一些開發(fā)中,可能會遇到需要長時間處理的任務(wù),此時就需要使用異步的方式來實(shí)現(xiàn),讓長時間任務(wù)在后臺運(yùn)行,先將本次請求的響應(yīng)狀態(tài)返回給前端,不讓前端界面「卡頓」,當(dāng)異步任務(wù)處理好后,如果需要返回狀態(tài),再將狀態(tài)返回。那么Flask怎么實(shí)現(xiàn)異步執(zhí)行任務(wù)呢?接下來的這篇文章帶你了解。
怎么實(shí)現(xiàn)呢?
使用線程的方式
當(dāng)要執(zhí)行耗時任務(wù)時,直接開啟一個新的線程來執(zhí)行任務(wù),這種方式最為簡單快速。
通過 ThreadPoolExecutor 來實(shí)現(xiàn)
from flask import Flask
from time import sleep
from concurrent.futures import ThreadPoolExecutor
# DOCS https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.ThreadPoolExecutor
# 創(chuàng)建線程池執(zhí)行器
executor = ThreadPoolExecutor(2)
app = Flask(__name__)
@app.route('/jobs')
def run_jobs():
# 交由線程去執(zhí)行耗時任務(wù)
executor.submit(long_task, 'hello', 123)
return 'long task running.'
# 耗時任務(wù)
def long_task(arg1, arg2):
print("args: %s %s!" % (arg1, arg2))
sleep(5)
print("Task is done!")
if __name__ == '__main__':
app.run()
當(dāng)要執(zhí)行一些比較簡單的耗時任務(wù)時就可以使用這種方式,如發(fā)郵件、發(fā)短信驗(yàn)證碼等。
但這種方式有個問題,就是前端無法得知任務(wù)執(zhí)行狀態(tài)。
如果想要前端知道,就需要設(shè)計(jì)一些邏輯,比如將任務(wù)執(zhí)行狀態(tài)存儲到 redis 中,通過唯一的任務(wù) id 進(jìn)行標(biāo)識,然后再寫一個接口,通過任務(wù) id 去獲取任務(wù)的狀態(tài),然后讓前端定時去請求該接口,從而獲得任務(wù)狀態(tài)信息。
全部自己實(shí)現(xiàn)就顯得有些麻煩了,而 Celery 剛好實(shí)現(xiàn)了這樣的邏輯,來使用一下。
使用 Celery
為了滿足前端可以獲得任務(wù)狀態(tài)的需求,可以使用 Celery。
Celery 是實(shí)時任務(wù)處理與調(diào)度的分布式任務(wù)隊(duì)列,它常用于 web 異步任務(wù)、定時任務(wù)等,后面單獨(dú)寫一篇文章描述 Celery 的架構(gòu),這里不深入討論。
現(xiàn)在我想讓前端可以通過一個進(jìn)度條來判斷后端任務(wù)的執(zhí)行情況。使用 Celery 就很容易實(shí)現(xiàn),首先通過 pip 安裝 Celery 與 redis,之所以要安裝 redis,是因?yàn)樽?Celery 選擇 redis 作為「消息代理 / 消息中間件」。
pip install celery pip install redis
在 Flask 中使用 Celery 其實(shí)很簡單,這里先簡單的過一下 Flask 中使用 Celery 的整體流程,然后再去實(shí)現(xiàn)具體的項(xiàng)目
1.在 Flask 中初始化 Celery
from flask import Flask
from celery import Celery
app = Flask(__name__)
# 配置
# 配置消息代理的路徑,如果是在遠(yuǎn)程服務(wù)器上,則配置遠(yuǎn)程服務(wù)器中redis的URL
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
# 要存儲 Celery 任務(wù)的狀態(tài)或運(yùn)行結(jié)果時就必須要配置
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'
# 初始化Celery
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
# 將Flask中的配置直接傳遞給Celery
celery.conf.update(app.config)
上述代碼中,通過 Celery 類初始化 celery 對象,傳入的應(yīng)用名稱與消息代理的連接 URL。
2.通過 celery.task 裝飾器裝飾耗時任務(wù)對應(yīng)的函數(shù)
@celery.task
def long_task(arg1, arg2):
# 耗時任務(wù)的邏輯
return result
3.Flask 中定義接口通過異步的方式執(zhí)行耗時任務(wù)
@app.route('/', methods=['GET', 'POST'])
def index():
task = long_task.delay(1, 2)
delay () 方法是 applyasync () 方法的快捷方式,applyasync () 參數(shù)更多,可以更加細(xì)致的控制耗時任務(wù),比如想要 long_task () 在一分鐘后再執(zhí)行
@app.route('/', methods=['GET', 'POST'])
def index():
task = long_task.apply_async(args=[1, 2], countdown=60)
delay () 與 apply_async () 會返回一個任務(wù)對象,該對象可以獲取任務(wù)的狀態(tài)與各種相關(guān)信息。
通過這 3 步就可以使用 Celery 了。
接著就具體來實(shí)現(xiàn)「讓前端可以通過一個進(jìn)度條來判斷后端任務(wù)的執(zhí)行情況」的需求。
# bind為True,會傳入self給被裝飾的方法
@celery.task(bind=True)
def long_task(self):
verb = ['Starting up', 'Booting', 'Repairing', 'Loading', 'Checking']
adjective = ['master', 'radiant', 'silent', 'harmonic', 'fast']
noun = ['solar array', 'particle reshaper', 'cosmic ray', 'orbiter', 'bit']
message = ''
total = random.randint(10, 50)
for i in range(total):
if not message or random.random() < 0.25:
# 隨機(jī)的獲取一些信息
message = '{0} {1} {2}...'.format(random.choice(verb),
random.choice(adjective),
random.choice(noun))
# 更新Celery任務(wù)狀態(tài)
self.update_state(state='PROGRESS',
meta={'current': i, 'total': total,
'status': message})
time.sleep(1)
# 返回字典
return {'current': 100, 'total': 100, 'status': 'Task completed!',
'result': 42}
上述代碼中,celery.task () 裝飾器使用了 bind=True 參數(shù),這個參數(shù)會讓 Celery 將 Celery 本身傳入,可以用于記錄與更新任務(wù)狀態(tài)。
然后就是一個 for 迭代,迭代的邏輯沒什么意義,就是隨機(jī)從 list 中抽取一些詞匯來模擬一些邏輯的運(yùn)行,為了表示這是耗時邏輯,通過 time.sleep (1) 休眠一秒。
每次獲取一次詞匯,就通過 self.update_state () 更新 Celery 任務(wù)的狀態(tài),Celery 包含一些內(nèi)置狀態(tài),如 SUCCESS、STARTED 等等,這里使用了自定義狀態(tài)「PROGRESS」,除了狀態(tài)外,還將本次循環(huán)的一些信息通過 meta 參數(shù) (元數(shù)據(jù)) 以字典的形式存儲起來。有了這些數(shù)據(jù),前端就可以顯示進(jìn)度條了。
定義好耗時方法后,再定義一個 Flask 接口方法來調(diào)用該耗時方法
@app.route('/longtask', methods=['POST'])
def longtask():
# 異步調(diào)用
task = long_task.apply_async()
# 返回 202,與Location頭
return jsonify({}), 202, {'Location': url_for('taskstatus',
task_id=task.id)}
簡單而言,前端通過 POST 請求到 /longtask,讓后端開始去執(zhí)行耗時任務(wù)。
返回的狀態(tài)碼為 202,202 通常表示一個請求正在進(jìn)行中,然后還在返回?cái)?shù)據(jù)包的包頭 (Header) 中添加了 Location 頭信息,前端可以通過讀取數(shù)據(jù)包中 Header 中的 Location 的信息來獲取任務(wù) id 對應(yīng)的完整 url。
前端有了任務(wù) id 對應(yīng)的 url 后,還需要提供一個接口給前端,讓前端可以通過任務(wù) id 去獲取當(dāng)前時刻任務(wù)的具體狀態(tài)。
@app.route('/status/<task_id>')
def taskstatus(task_id):
task = long_task.AsyncResult(task_id)
if task.state == 'PENDING': # 在等待
response = {
'state': task.state,
'current': 0,
'total': 1,
'status': 'Pending...'
}
elif task.state != 'FAILURE': # 沒有失敗
response = {
'state': task.state, # 狀態(tài)
# meta中的數(shù)據(jù),通過task.info.get()可以獲得
'current': task.info.get('current', 0), # 當(dāng)前循環(huán)進(jìn)度
'total': task.info.get('total', 1), # 總循環(huán)進(jìn)度
'status': task.info.get('status', '')
}
if 'result' in task.info:
response['result'] = task.info['result']
else:
# 后端執(zhí)行任務(wù)出現(xiàn)了一些問題
response = {
'state': task.state,
'current': 1,
'total': 1,
'status': str(task.info), # 報錯的具體異常
}
return jsonify(response)
為了可以獲得任務(wù)對象中的信息,使用任務(wù) id 初始化 AsyncResult 類,獲得任務(wù)對象,然后就可以從任務(wù)對象中獲得當(dāng)前任務(wù)的信息。
該方法會返回一個 JSON,其中包含了任務(wù)狀態(tài)以及 meta 中指定的信息,前端可以利用這些信息構(gòu)建一個進(jìn)度條。
如果任務(wù)在 PENDING 狀態(tài),表示該任務(wù)還沒有開始,在這種狀態(tài)下,任務(wù)中是沒有什么信息的,這里人為的返回一些數(shù)據(jù)。如果任務(wù)執(zhí)行失敗,就返回 task.info 中包含的異常信息,此外就是正常執(zhí)行了,正常執(zhí)行可以通 task.info 獲得任務(wù)中具體的信息。
這樣,后端的邏輯就處理完成了,接著就來實(shí)現(xiàn)前端的邏輯,要實(shí)現(xiàn)圖形進(jìn)度條,可以直接使用 nanobar.js,簡單兩句話就可以實(shí)現(xiàn)一個進(jìn)度條,其官網(wǎng)例子如下:
var options = {
classname: 'my-class',
id: 'my-id',
// 進(jìn)度條要出現(xiàn)的位置
target: document.getElementById('myDivId')
};
// 初始化進(jìn)度條對象
var nanobar = new Nanobar( options );
nanobar.go( 30 ); // 30% 進(jìn)度條
nanobar.go( 76 ); // 76% 進(jìn)度條
// 100% 進(jìn)度條,進(jìn)度條結(jié)束
nanobar.go(100);
有了 nanobar.js 就非常簡單了。
先定義一個簡單的 HTML 界面
<h2>Long running task with progress updates</h2>
<button id="start-bg-job">Start Long Calculation</button><br><br>
<div id="progress"></div>
通過 JavaScript 實(shí)現(xiàn)對后臺的請求
// 按鈕點(diǎn)擊事件
$(function() {
$('#start-bg-job').click(start_long_task);
});
// 請求 longtask 接口
function start_long_task() {
// 添加元素在html中
div = $('<div class="progress"><div></div><div>0%</div><div>...</div><div> </div></div><hr>');
$('#progress').append(div);
// 創(chuàng)建進(jìn)度條對象
var nanobar = new Nanobar({
bg: '#44f',
target: div[0].childNodes[0]
});
// ajax請求longtask
$.ajax({
type: 'POST',
url: '/longtask',
// 獲得數(shù)據(jù),從響應(yīng)頭中獲取Location
success: function(data, status, request) {
status_url = request.getResponseHeader('Location');
// 調(diào)用 update_progress() 方法更新進(jìn)度條
update_progress(status_url, nanobar, div[0]);
},
error: function() {
alert('Unexpected error');
}
});
}
// 更新進(jìn)度條
function update_progress(status_url, nanobar, status_div) {
// getJSON()方法是JQuery內(nèi)置方法,這里向Location中對應(yīng)的url發(fā)起請求,即請求「/status/<task_id>」
$.getJSON(status_url, function(data) {
// 計(jì)算進(jìn)度
percent = parseInt(data['current'] * 100 / data['total']);
// 更新進(jìn)度條
nanobar.go(percent);
// 更新文字
$(status_div.childNodes[1]).text(percent + '%');
$(status_div.childNodes[2]).text(data['status']);
if (data['state'] != 'PENDING' && data['state'] != 'PROGRESS') {
if ('result' in data) {
// 展示結(jié)果
$(status_div.childNodes[3]).text('Result: ' + data['result']);
}
else {
// 意料之外的事情發(fā)生
$(status_div.childNodes[3]).text('Result: ' + data['state']);
}
}
else {
// 2秒后再次運(yùn)行
setTimeout(function() {
update_progress(status_url, nanobar, status_div);
}, 2000);
}
});
}
可以通過注釋閱讀代碼整體邏輯。
至此,需求實(shí)現(xiàn)完了,運(yùn)行一下。
首先運(yùn)行 Redis
redis-server
然后運(yùn)行 celery
celery worker -A app.celery --loglevel=info
最后運(yùn)行 Flask 項(xiàng)目
python app.py
效果如下:
到此這篇Flask怎么實(shí)現(xiàn)異步執(zhí)行任務(wù)的文章就介紹到這了,更多Flask 學(xué)習(xí)內(nèi)容請搜索W3Cschool以前的文章或繼續(xù)瀏覽下面的相關(guān)文章。