原文鏈接： https://www.tinymind.cn/competitions/48
（▲由Python大本營付費下載自視覺中國）
作者 | ayuliao ?
出自 | hackpython（ID：hackpython）
出自 | hackpython（ID：hackpython）
簡介
Flask 是 Python 中有名的輕量級同步 web 框架，在一些開發(fā)中，可能會遇到需要長時間處理的任務，此時就需要使用異步的方式來實現(xiàn)，讓長時間任務在后臺運行，先將本次請求的響應狀態(tài)返回給前端，不讓前端界面「卡頓」，當異步任務處理好后，如果需要返回狀態(tài)，再將狀態(tài)返回。
怎么實現(xiàn)呢？
使用線程的方式
當要執(zhí)行耗時任務時，直接開啟一個新的線程來執(zhí)行任務，這種方式最為簡單快速。
通過 ThreadPoolExecutor 來實現(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__)
def ? run_jobs () :
???? #?交由線程去執(zhí)行耗時任務
????executor.submit(long_task,? 'hello' ,? 123 )
???? return ? 'long?task?running.'
#?耗時任務
def ? long_task (arg1,?arg2) :
????print( "args:?%s?%s!" ?%?(arg1,?arg2))
????sleep( 5 )
????print( "Task?is?done!" )
if ?__name__?==? '__main__' :
????app.run()
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__)
def ? run_jobs () :
???? #?交由線程去執(zhí)行耗時任務
????executor.submit(long_task,? 'hello' ,? 123 )
???? return ? 'long?task?running.'
#?耗時任務
def ? long_task (arg1,?arg2) :
????print( "args:?%s?%s!" ?%?(arg1,?arg2))
????sleep( 5 )
????print( "Task?is?done!" )
if ?__name__?==? '__main__' :
????app.run()
當要執(zhí)行一些比較簡單的耗時任務時就可以使用這種方式，如發(fā)郵件、發(fā)短信驗證碼等。
但這種方式有個問題，就是前端無法得知任務執(zhí)行狀態(tài)。
如果想要前端知道，就需要設計一些邏輯，比如將任務執(zhí)行狀態(tài)存儲到 redis 中，通過唯一的任務 id 進行標識，然后再寫一個接口，通過任務 id 去獲取任務的狀態(tài)，然后讓前端定時去請求該接口，從而獲得任務狀態(tài)信息。
全部自己實現(xiàn)就顯得有些麻煩了，而 Celery 剛好實現(xiàn)了這樣的邏輯，來使用一下。
使用 Celery
為了滿足前端可以獲得任務狀態(tài)的需求，可以使用 Celery。
Celery 是實時任務處理與調(diào)度的分布式任務隊列，它常用于 web 異步任務、定時任務等，后面單獨寫一篇文章描述 Celery 的架構(gòu)，這里不深入討論。
現(xiàn)在我想讓前端可以通過一個進度條來判斷后端任務的執(zhí)行情況。使用 Celery 就很容易實現(xiàn)，首先通過 pip 安裝 Celery 與 redis，之所以要安裝 redis，是因為讓 Celery 選擇 redis 作為「消息代理 / 消息中間件」。
pip?
install
?celery
pip? install ?redis
pip? install ?redis
在 Flask 中使用 Celery 其實很簡單，這里先簡單的過一下 Flask 中使用 Celery 的整體流程，然后再去實現(xiàn)具體的項目
1.在 Flask 中初始化 Celery
from
?flask?
import
?Flask
from ?celery? import ?Celery
app?=?Flask(__name__)
#?配置
#?配置消息代理的路徑，如果是在遠程服務器上，則配置遠程服務器中redis的URL
app.config[ 'CELERY_BROKER_URL' ]?=? 'redis://localhost:6379/0'
#?要存儲?Celery?任務的狀態(tài)或運行結(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)
from ?celery? import ?Celery
app?=?Flask(__name__)
#?配置
#?配置消息代理的路徑，如果是在遠程服務器上，則配置遠程服務器中redis的URL
app.config[ 'CELERY_BROKER_URL' ]?=? 'redis://localhost:6379/0'
#?要存儲?Celery?任務的狀態(tài)或運行結(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 對象，傳入的應用名稱與消息代理的連接 URL。
2.通過 celery.task 裝飾器裝飾耗時任務對應的函數(shù)
def ? long_task (arg1,?arg2) :
???? #?耗時任務的邏輯
???? return ?result
3.Flask 中定義接口通過異步的方式執(zhí)行耗時任務
@app.route(
'/'
,?methods=[
'GET'
,?
'POST'
])
def ? index () :
????task?=?long_task.delay( 1 ,? 2 )
delay?()?方法是?applyasync?()?方法的快捷方式，applyasync?()?參數(shù)更多，可以更加細致的控制耗時任務，比如想要?long_task?()?在一分鐘后再執(zhí)行
@app.route( '/' ,?methods=[ 'GET' ,? 'POST' ])
def ? index () :
????task?=?long_task.apply_async(args=[ 1 ,? 2 ],?countdown= 60 )
delay?()?與?apply_async?()?會返回一個任務對象，該對象可以獲取任務的狀態(tài)與各種相關(guān)信息。
def ? index () :
????task?=?long_task.delay( 1 ,? 2 )
delay?()?方法是?applyasync?()?方法的快捷方式，applyasync?()?參數(shù)更多，可以更加細致的控制耗時任務，比如想要?long_task?()?在一分鐘后再執(zhí)行
@app.route( '/' ,?methods=[ 'GET' ,? 'POST' ])
def ? index () :
????task?=?long_task.apply_async(args=[ 1 ,? 2 ],?countdown= 60 )
delay?()?與?apply_async?()?會返回一個任務對象，該對象可以獲取任務的狀態(tài)與各種相關(guān)信息。
通過這 3 步就可以使用 Celery 了。
接著就具體來實現(xiàn)「讓前端可以通過一個進度條來判斷后端任務的執(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 :
???????????? #?隨機的獲取一些信息
????????????message?=? '{0}?{1}?{2}...' .format(random.choice(verb),
??????????????????????????????????????????????random.choice(adjective),
??????????????????????????????????????????????random.choice(noun))
???????? #?更新Celery任務狀態(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)
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 :
???????????? #?隨機的獲取一些信息
????????????message?=? '{0}?{1}?{2}...' .format(random.choice(verb),
??????????????????????????????????????????????random.choice(adjective),
??????????????????????????????????????????????random.choice(noun))
???????? #?更新Celery任務狀態(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 本身傳入，可以用于記錄與更新任務狀態(tài)。
然后就是一個 for 迭代，迭代的邏輯沒什么意義，就是隨機從 list 中抽取一些詞匯來模擬一些邏輯的運行，為了表示這是耗時邏輯，通過 time.sleep (1) 休眠一秒。
每次獲取一次詞匯，就通過 self.update_state () 更新 Celery 任務的狀態(tài)，Celery 包含一些內(nèi)置狀態(tài)，如 SUCCESS、STARTED 等等，這里使用了自定義狀態(tài)「PROGRESS」，除了狀態(tài)外，還將本次循環(huán)的一些信息通過 meta 參數(shù) (元數(shù)據(jù)) 以字典的形式存儲起來。有了這些數(shù)據(jù)，前端就可以顯示進度條了。
定義好耗時方法后，再定義一個 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)}
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í)行耗時任務。
返回的狀態(tài)碼為 202，202 通常表示一個請求正在進行中，然后還在返回數(shù)據(jù)包的包頭 (Header) 中添加了 Location 頭信息，前端可以通過讀取數(shù)據(jù)包中 Header 中的 Location 的信息來獲取任務 id 對應的完整 url。
前端有了任務 id 對應的 url 后，還需要提供一個接口給前端，讓前端可以通過任務 id 去獲取當前時刻任務的具體狀態(tài)。
@app.route(
'/status/
'
)
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,?
????????????
???????????? 'current' :?task.info. get ( 'current' ,? 0 ),?
???????????? 'total' :?task.info. get ( 'total' ,? 1 ),?
???????????? 'status' :?task.info. get ( 'status' ,? '' )
????????}
???????? if ? 'result' ? in ?task.info:
????????????response[ 'result' ]?=?task.info[ 'result' ]
???? else :
????????
????????response?=?{
???????????? 'state' :?task.state,
???????????? 'current' :? 1 ,
???????????? 'total' :? 1 ,
???????????? 'status' :?str(task.info),??
????????}
???? return ?jsonify(response)
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,?
????????????
???????????? 'current' :?task.info. get ( 'current' ,? 0 ),?
???????????? 'total' :?task.info. get ( 'total' ,? 1 ),?
???????????? 'status' :?task.info. get ( 'status' ,? '' )
????????}
???????? if ? 'result' ? in ?task.info:
????????????response[ 'result' ]?=?task.info[ 'result' ]
???? else :
????????
????????response?=?{
???????????? 'state' :?task.state,
???????????? 'current' :? 1 ,
???????????? 'total' :? 1 ,
???????????? 'status' :?str(task.info),??
????????}
???? return ?jsonify(response)
為了可以獲得任務對象中的信息，使用任務 id 初始化 AsyncResult 類，獲得任務對象，然后就可以從任務對象中獲得當前任務的信息。
該方法會返回一個 JSON，其中包含了任務狀態(tài)以及 meta 中指定的信息，前端可以利用這些信息構(gòu)建一個進度條。
如果任務在 PENDING 狀態(tài)，表示該任務還沒有開始，在這種狀態(tài)下，任務中是沒有什么信息的，這里人為的返回一些數(shù)據(jù)。如果任務執(zhí)行失敗，就返回 task.info 中包含的異常信息，此外就是正常執(zhí)行了，正常執(zhí)行可以通 task.info 獲得任務中具體的信息。
這樣，后端的邏輯就處理完成了，接著就來實現(xiàn)前端的邏輯，要實現(xiàn)圖形進度條，可以直接使用 nanobar.js，簡單兩句話就可以實現(xiàn)一個進度條，其官網(wǎng)例子如下：
var
?options?=?{
????classname:? 'my-class' ,
??id:? 'my-id' ,
?? //?進度條要出現(xiàn)的位置
????target:?document.getElementById( 'myDivId' )
};
//?初始化進度條對象
var ?nanobar?=? new ?Nanobar(?options?);
nanobar. go (? 30 ?);? //?30%?進度條
nanobar. go (? 76 ?);? //?76%?進度條
//?100%?進度條，進度條結(jié)束
nanobar. go ( 100 );
????classname:? 'my-class' ,
??id:? 'my-id' ,
?? //?進度條要出現(xiàn)的位置
????target:?document.getElementById( 'myDivId' )
};
//?初始化進度條對象
var ?nanobar?=? new ?Nanobar(?options?);
nanobar. go (? 30 ?);? //?30%?進度條
nanobar. go (? 76 ?);? //?76%?進度條
//?100%?進度條，進度條結(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 >
< button ? id = "start-bg-job" > Start?Long?Calculation button >< br >< br >
< div ? id = "progress" >div >
通過 JavaScript 實現(xiàn)對后臺的請求
//?按鈕點擊事件
$( function ( )? {
???????$( '#start-bg-job' ).click(start_long_task);
???});
//?請求?longtask?接口
function ? start_long_task ( )? {
???????????? //?添加元素在html中
????????????div?=?$( '
' );
????????????$( '#progress' ).append(div);
???????????? //?創(chuàng)建進度條對象
???????????? var ?nanobar?=? new ?Nanobar({
???????????????? bg :? '#44f' ,
???????????????? target :?div[ 0 ].childNodes[ 0 ]
????????????});
???????????? //?ajax請求longtask
????????????$.ajax({
???????????????? type :? 'POST' ,
???????????????? url :? '/longtask' ,
???????????????? //?獲得數(shù)據(jù)，從響應頭中獲取Location
????????????????success:? function ( data,?status,?request )? {
????????????????????status_url?=?request.getResponseHeader( 'Location' );
???????????????????? //?調(diào)用?update_progress()?方法更新進度條
????????????????????update_progress(status_url,?nanobar,?div[ 0 ]);
????????????????},
???????????????? error :? function ( )? {
????????????????????alert( 'Unexpected?error' );
????????????????}
????????????});
????????}
//?更新進度條
function ? update_progress ( status_url,?nanobar,?status_div )? {
???????????? //?getJSON()方法是JQuery內(nèi)置方法，這里向Location中對應的url發(fā)起請求，即請求「/status/
」
????????????$.getJSON(status_url,? function ( data )? {
???????????????? //?計算進度
????????????????percent?=? parseInt (data[ 'current' ]?*? 100 ?/?data[ 'total' ]);
???????????????? //?更新進度條
????????????????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秒后再次運行
????????????????????setTimeout( function ( )? {
????????????????????????update_progress(status_url,?nanobar,?status_div);
????????????????????},? 2000 );
????????????????}
????????????});?
????????}???????
$( function ( )? {
???????$( '#start-bg-job' ).click(start_long_task);
???});
//?請求?longtask?接口
function ? start_long_task ( )? {
???????????? //?添加元素在html中
????????????div?=?$( '
0%
...
?
' );
????????????$( '#progress' ).append(div);
???????????? //?創(chuàng)建進度條對象
???????????? var ?nanobar?=? new ?Nanobar({
???????????????? bg :? '#44f' ,
???????????????? target :?div[ 0 ].childNodes[ 0 ]
????????????});
???????????? //?ajax請求longtask
????????????$.ajax({
???????????????? type :? 'POST' ,
???????????????? url :? '/longtask' ,
???????????????? //?獲得數(shù)據(jù)，從響應頭中獲取Location
????????????????success:? function ( data,?status,?request )? {
????????????????????status_url?=?request.getResponseHeader( 'Location' );
???????????????????? //?調(diào)用?update_progress()?方法更新進度條
????????????????????update_progress(status_url,?nanobar,?div[ 0 ]);
????????????????},
???????????????? error :? function ( )? {
????????????????????alert( 'Unexpected?error' );
????????????????}
????????????});
????????}
//?更新進度條
function ? update_progress ( status_url,?nanobar,?status_div )? {
???????????? //?getJSON()方法是JQuery內(nèi)置方法，這里向Location中對應的url發(fā)起請求，即請求「/status/
????????????$.getJSON(status_url,? function ( data )? {
???????????????? //?計算進度
????????????????percent?=? parseInt (data[ 'current' ]?*? 100 ?/?data[ 'total' ]);
???????????????? //?更新進度條
????????????????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秒后再次運行
????????????????????setTimeout( function ( )? {
????????????????????????update_progress(status_url,?nanobar,?status_div);
????????????????????},? 2000 );
????????????????}
????????????});?
????????}???????
可以通過注釋閱讀代碼整體邏輯。
至此，需求實現(xiàn)完了，運行一下。
首先運行 Redis
redis-
server
然后運行 celery
celery
?worker?-A?app.celery?--loglevel=
info
最后運行 Flask 項目
python
?
app
.py
效果如下：
尾
Flask 異步運行任務的常見方式就介紹完了，因為本人在開發(fā)一個用于自動生成字幕的小玩具，其中視頻上傳以及字幕生成都是耗時任務，所以就單獨寫一篇文章來介紹一下這部分的內(nèi)容，后面會將小玩具的代碼開源讓大家學習一下，先一睹其真容：
（*本文為Python大本營轉(zhuǎn)載文章，轉(zhuǎn)載請聯(lián)系原作者）
◆
精彩推薦
◆
【結(jié)果提交倒計時】PV,UV流量預測算法大賽，結(jié)果提交截止時間為
9月20日
，還沒有提交的小伙伴抓緊時間了~~9月25日公布初賽成績。
最新排行榜請點擊
閱讀原文
查看。
推薦閱讀
-
-
5大必知的圖算法，附Python代碼實現(xiàn)
-
-
如何用爬蟲技術(shù)幫助孩子秒到心儀的幼兒園（基礎(chǔ)篇）
-
-
2019年最新華為、BAT、美團、頭條、滴滴面試題目及答案匯總
-
阿里巴巴楊群：高并發(fā)場景下Python的性能挑戰(zhàn)
你點的每個“在看”，我都認真當成了喜歡
更多文章、技術(shù)交流、商務合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號聯(lián)系： 360901061
您的支持是博主寫作最大的動力，如果您喜歡我的文章，感覺我的文章對您有幫助，請用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點擊下面給點支持吧，站長非常感激您！手機微信長按不能支付解決辦法：請將微信支付二維碼保存到相冊，切換到微信，然后點擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對您有幫助就好】元
