|
來源：人世間 www.jianshu.com/p/b5e347b3a17c 如有好文章投稿,，請點擊 → 這里了解詳情
python asyncio
網(wǎng)絡(luò)模型有很多中,，為了實現(xiàn)高并發(fā)也有很多方案，多線程,，多進程,。無論多線程和多進程,，IO的調(diào)度更多取決于系統(tǒng),，而協(xié)程的方式，調(diào)度來自用戶,，用戶可以在函數(shù)中yield一個狀態(tài),。使用協(xié)程可以實現(xiàn)高效的并發(fā)任務(wù)。Python的在3.4中引入了協(xié)程的概念,，可是這個還是以生成器對象為基礎(chǔ),，3.5則確定了協(xié)程的語法。下面將簡單介紹asyncio的使用,。實現(xiàn)協(xié)程的不僅僅是asyncio，tornado和gevent都實現(xiàn)了類似的功能,。
event_loop 事件循環(huán)：程序開啟一個無限的循環(huán),，程序員會把一些函數(shù)注冊到事件循環(huán)上。當(dāng)滿足事件發(fā)生的時候,，調(diào)用相應(yīng)的協(xié)程函數(shù),。 coroutine 協(xié)程：協(xié)程對象，指一個使用async關(guān)鍵字定義的函數(shù),，它的調(diào)用不會立即執(zhí)行函數(shù),，而是會返回一個協(xié)程對象。協(xié)程對象需要注冊到事件循環(huán),，由事件循環(huán)調(diào)用,。 task 任務(wù)：一個協(xié)程對象就是一個原生可以掛起的函數(shù)，任務(wù)則是對協(xié)程進一步封裝,，其中包含任務(wù)的各種狀態(tài),。 future： 代表將來執(zhí)行或沒有執(zhí)行的任務(wù)的結(jié)果。它和task上沒有本質(zhì)的區(qū)別 async/await 關(guān)鍵字：python3.5 用于定義協(xié)程的關(guān)鍵字,，async定義一個協(xié)程,，await用于掛起阻塞的異步調(diào)用接口。
上述的概念單獨拎出來都不好懂,，比較他們之間是相互聯(lián)系,，一起工作。下面看例子,，再回溯上述概念,，更利于理解。
定義一個協(xié)程
定義一個協(xié)程很簡單,，使用async關(guān)鍵字,，就像定義普通函數(shù)一樣：
import time import asyncio now = lambda : time.time() async def do_some_work(x): print('Waiting: ', x) start = now() coroutine = do_some_work(2) loop = asyncio.get_event_loop() loop.run_until_complete(coroutine) print('TIME: ', now() - start)
通過async關(guān)鍵字定義一個協(xié)程（coroutine）,，協(xié)程也是一種對象。協(xié)程不能直接運行,，需要把協(xié)程加入到事件循環(huán)（loop）,，由后者在適當(dāng)?shù)臅r候調(diào)用協(xié)程。asyncio.get_event_loop方法可以創(chuàng)建一個事件循環(huán),，然后使用run_until_complete將協(xié)程注冊到事件循環(huán),，并啟動事件循環(huán)。因為本例只有一個協(xié)程,，于是可以看見如下輸出：
Waiting: 2 TIME: 0.0004658699035644531
創(chuàng)建一個task
協(xié)程對象不能直接運行,，在注冊事件循環(huán)的時候，其實是run_until_complete方法將協(xié)程包裝成為了一個任務(wù)（task）對象,。所謂task對象是Future類的子類,。保存了協(xié)程運行后的狀態(tài)，用于未來獲取協(xié)程的結(jié)果,。
import asyncio import time now = lambda : time.time() async def do_some_work(x): print('Waiting: ', x) start = now() coroutine = do_some_work(2) loop = asyncio.get_event_loop() # task = asyncio.ensure_future(coroutine) task = loop.create_task(coroutine) print(task) loop.run_until_complete(task) print(task) print('TIME: ', now() - start)
可以看到輸出結(jié)果為：
Task pending coro=do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:17>> Waiting: 2 Task finished coro=do_some_work() done, defined at /Users/ghost/Rsj217/python3.6/async/async-main.py:17> result=None> TIME: 0.0003490447998046875
創(chuàng)建task后,，task在加入事件循環(huán)之前是pending狀態(tài)，因為do_some_work中沒有耗時的阻塞操作,，task很快就執(zhí)行完畢了,。后面打印的finished狀態(tài)。
asyncio.ensure_future(coroutine) 和 loop.create_task(coroutine)都可以創(chuàng)建一個task,，run_until_complete的參數(shù)是一個futrue對象,。當(dāng)傳入一個協(xié)程，其內(nèi)部會自動封裝成task,，task是Future的子類,。isinstance(task, asyncio.Future)將會輸出True。
綁定回調(diào)
綁定回調(diào),，在task執(zhí)行完畢的時候可以獲取執(zhí)行的結(jié)果,，回調(diào)的最后一個參數(shù)是future對象，通過該對象可以獲取協(xié)程返回值,。如果回調(diào)需要多個參數(shù),，可以通過偏函數(shù)導(dǎo)入。
import time import asyncio now = lambda : time.time() async def do_some_work(x): print('Waiting: ', x) return 'Done after {}s'.format(x) def callback(future): print('Callback: ', future.result()) start = now() coroutine = do_some_work(2) loop = asyncio.get_event_loop() task = asyncio.ensure_future(coroutine) task.add_done_callback(callback) loop.run_until_complete(task) print('TIME: ', now() - start)
def callback(t, future): print('Callback:', t, future.result()) task.add_done_callback(functools.partial(callback, 2))
可以看到,，coroutine執(zhí)行結(jié)束時候會調(diào)用回調(diào)函數(shù),。并通過參數(shù)future獲取協(xié)程執(zhí)行的結(jié)果。我們創(chuàng)建的task和回調(diào)里的future對象,，實際上是同一個對象,。
future 與 result
回調(diào)一直是很多異步編程的惡夢，程序員更喜歡使用同步的編寫方式寫異步代碼,，以避免回調(diào)的惡夢,?；卣{(diào)中我們使用了future對象的result方法。前面不綁定回調(diào)的例子中,，我們可以看到task有fiinished狀態(tài),。在那個時候，可以直接讀取task的result方法,。
async def do_some_work(x): print('Waiting {}'.format(x)) return 'Done after {}s'.format(x) start = now() coroutine = do_some_work(2) loop = asyncio.get_event_loop() task = asyncio.ensure_future(coroutine) loop.run_until_complete(task) print('Task ret: {}'.format(task.result())) print('TIME: {}'.format(now() - start))
可以看到輸出的結(jié)果：
Waiting: 2 Task ret: Done after 2s TIME: 0.0003650188446044922
阻塞和await
使用async可以定義協(xié)程對象,，使用await可以針對耗時的操作進行掛起，就像生成器里的yield一樣,，函數(shù)讓出控制權(quán),。協(xié)程遇到await，事件循環(huán)將會掛起該協(xié)程,，執(zhí)行別的協(xié)程,，直到其他的協(xié)程也掛起或者執(zhí)行完畢，再進行下一個協(xié)程的執(zhí)行,。
耗時的操作一般是一些IO操作,，例如網(wǎng)絡(luò)請求，文件讀取等,。我們使用asyncio.sleep函數(shù)來模擬IO操作。協(xié)程的目的也是讓這些IO操作異步化,。
import asyncio import time now = lambda: time.time() async def do_some_work(x): print('Waiting: ', x) await asyncio.sleep(x) return 'Done after {}s'.format(x) start = now() coroutine = do_some_work(2) loop = asyncio.get_event_loop() task = asyncio.ensure_future(coroutine) loop.run_until_complete(task) print('Task ret: ', task.result()) print('TIME: ', now() - start)
在 sleep的時候,，使用await讓出控制權(quán)。即當(dāng)遇到阻塞調(diào)用的函數(shù)的時候,，使用await方法將協(xié)程的控制權(quán)讓出,，以便loop調(diào)用其他的協(xié)程。現(xiàn)在我們的例子就用耗時的阻塞操作了,。
并發(fā)和并行
并發(fā)和并行一直是容易混淆的概念,。并發(fā)通常指有多個任務(wù)需要同時進行，并行則是同一時刻有多個任務(wù)執(zhí)行,。用上課來舉例就是,，并發(fā)情況下是一個老師在同一時間段輔助不同的人功課。并行則是好幾個老師分別同時輔助多個學(xué)生功課,。簡而言之就是一個人同時吃三個饅頭還是三個人同時分別吃一個的情況,，吃一個饅頭算一個任務(wù)。
asyncio實現(xiàn)并發(fā),，就需要多個協(xié)程來完成任務(wù),，每當(dāng)有任務(wù)阻塞的時候就await，然后其他協(xié)程繼續(xù)工作,。創(chuàng)建多個協(xié)程的列表,，然后將這些協(xié)程注冊到事件循環(huán)中,。
import asyncio import time now = lambda: time.time() async def do_some_work(x): print('Waiting: ', x) await asyncio.sleep(x) return 'Done after {}s'.format(x) start = now() coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(4) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) for task in tasks: print('Task ret: ', task.result()) print('TIME: ', now() - start)
結(jié)果如下
Waiting: 1 Waiting: 2 Waiting: 4 Task ret: Done after 1s Task ret: Done after 2s Task ret: Done after 4s TIME: 4.003541946411133
總時間為4s左右。4s的阻塞時間,，足夠前面兩個協(xié)程執(zhí)行完畢,。如果是同步順序的任務(wù)，那么至少需要7s,。此時我們使用了aysncio實現(xiàn)了并發(fā),。asyncio.wait(tasks) 也可以使用 asyncio.gather(*tasks) ,前者接受一個task列表，后者接收一堆task,。
協(xié)程嵌套
使用async可以定義協(xié)程,，協(xié)程用于耗時的io操作，我們也可以封裝更多的io操作過程,，這樣就實現(xiàn)了嵌套的協(xié)程,，即一個協(xié)程中await了另外一個協(xié)程，如此連接起來,。
import asyncio import time now = lambda: time.time() async def do_some_work(x): print('Waiting: ', x) await asyncio.sleep(x) return 'Done after {}s'.format(x) async def main(): coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(4) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] dones, pendings = await asyncio.wait(tasks) for task in dones: print('Task ret: ', task.result()) start = now() loop = asyncio.get_event_loop() loop.run_until_complete(main()) print('TIME: ', now() - start)
如果使用的是 asyncio.gather創(chuàng)建協(xié)程對象,，那么await的返回值就是協(xié)程運行的結(jié)果。
results = await asyncio.gather(*tasks) for result in results: print('Task ret: ', result)
不在main協(xié)程函數(shù)里處理結(jié)果,，直接返回await的內(nèi)容,，那么最外層的run_until_complete將會返回main協(xié)程的結(jié)果。
async def main(): coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(2) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] return await asyncio.gather(*tasks) start = now() loop = asyncio.get_event_loop() results = loop.run_until_complete(main()) for result in results: print('Task ret: ', result)
或者返回使用asyncio.wait方式掛起協(xié)程,。
async def main(): coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(4) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] return await asyncio.wait(tasks) start = now() loop = asyncio.get_event_loop() done, pending = loop.run_until_complete(main()) for task in done: print('Task ret: ', task.result())
也可以使用asyncio的as_completed方法
async def main(): coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(4) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] for task in asyncio.as_completed(tasks): result = await task print('Task ret: {}'.format(result)) start = now() loop = asyncio.get_event_loop() done = loop.run_until_complete(main()) print('TIME: ', now() - start)
由此可見,，協(xié)程的調(diào)用和組合十分靈活，尤其是對于結(jié)果的處理,，如何返回,，如何掛起，需要逐漸積累經(jīng)驗和前瞻的設(shè)計,。
協(xié)程停止
上面見識了協(xié)程的幾種常用的用法,，都是協(xié)程圍繞著事件循環(huán)進行的操作。future對象有幾個狀態(tài)：
Pending Running Done Cancelled
創(chuàng)建future的時候,，task為pending,，事件循環(huán)調(diào)用執(zhí)行的時候當(dāng)然就是running，調(diào)用完畢自然就是done,，如果需要停止事件循環(huán),，就需要先把task取消?？梢允褂胊syncio.Task獲取事件循環(huán)的task
import asyncio import time now = lambda: time.time() async def do_some_work(x): print('Waiting: ', x) await asyncio.sleep(x) return 'Done after {}s'.format(x) coroutine1 = do_some_work(1) coroutine2 = do_some_work(2) coroutine3 = do_some_work(2) tasks = [ asyncio.ensure_future(coroutine1), asyncio.ensure_future(coroutine2), asyncio.ensure_future(coroutine3) ] start = now() loop = asyncio.get_event_loop() try: loop.run_until_complete(asyncio.wait(tasks)) except KeyboardInterrupt as e: print(asyncio.Task.all_tasks()) for task in asyncio.Task.all_tasks(): print(task.cancel()) loop.stop() loop.run_forever() finally: loop.close() print('TIME: ', now() - start)
啟動事件循環(huán)之后,，馬上ctrl+c，會觸發(fā)run_until_complete的執(zhí)行異常 KeyBorardInterrupt,。然后通過循環(huán)asyncio.Task取消future,?？梢钥吹捷敵鋈缦拢?/p>
Waiting: 1 Waiting: 2 Waiting: 2 {Task pending coro=do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=Future pending cb=[TaskWakeupMethWrapper object at 0x101230648>()]> cb=[_wait.locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>, Task pending coro=do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=Future pending cb=[TaskWakeupMethWrapper object at 0x1032b10a8>()]> cb=[_wait.locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>, Task pending coro=wait() running at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:307> wait_for=Future pending cb=[TaskWakeupMethWrapper object at 0x103317d38>()]> cb=[_run_until_complete_cb() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/base_events.py:176]>, Task pending coro=do_some_work() running at /Users/ghost/Rsj217/python3.6/async/async-main.py:18> wait_for=Future pending cb=[TaskWakeupMethWrapper object at 0x103317be8>()]> cb=[_wait.locals>._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>} True True True True TIME: 0.8858370780944824
True表示cannel成功，loop stop之后還需要再次開啟事件循環(huán),，最后在close,，不然還會拋出異常：
Task was destroyed but it is pending! task: Task pending coro=do_some_work() done,
循環(huán)task，逐個cancel是一種方案,，可是正如上面我們把task的列表封裝在main函數(shù)中,，main函數(shù)外進行事件循環(huán)的調(diào)用。這個時候,，main相當(dāng)于最外出的一個task,，那么處理包裝的main函數(shù)即可。
不同線程的事件循環(huán)
很多時候,，我們的事件循環(huán)用于注冊協(xié)程,，而有的協(xié)程需要動態(tài)的添加到事件循環(huán)中。一個簡單的方式就是使用多線程,。當(dāng)前線程創(chuàng)建一個事件循環(huán),，然后在新建一個線程，在新線程中啟動事件循環(huán),。當(dāng)前線程不會被block,。
啟動上述代碼之后，當(dāng)前線程不會被block,，新線程中會按照順序執(zhí)行call_soon_threadsafe方法注冊的more_work方法,，后者因為time.sleep操作是同步阻塞的，因此運行完畢more_work需要大致6 + 3
新線程協(xié)程
上述的例子,，主線程中創(chuàng)建一個new_loop,，然后在另外的子線程中開啟一個無限事件循環(huán),。主線程通過run_coroutine_threadsafe新注冊協(xié)程對象,。這樣就能在子線程中進行事件循環(huán)的并發(fā)操作，同時主線程又不會被block,。一共執(zhí)行的時間大概在6s左右,。
master-worker主從模式
對于并發(fā)任務(wù)，通常是用生成消費模型,，對隊列的處理可以使用類似master-worker的方式,，master主要用戶獲取隊列的msg，worker用戶處理消息,。
為了簡單起見,，并且協(xié)程更適合單線程的方式，我們的主線程用來監(jiān)聽隊列,，子線程用于處理隊列,。這里使用redis的隊列,。主線程中有一個是無限循環(huán)，用戶消費隊列,。
while True: task = rcon.rpop('queue') if not task: time.sleep(1) continue asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
給隊列添加一些數(shù)據(jù)：
可以看見輸出：
我們發(fā)起了一個耗時5s的操作,，然后又發(fā)起了連個1s的操作，可以看見子線程并發(fā)的執(zhí)行了這幾個任務(wù),，其中5s awati的時候,，相繼執(zhí)行了1s的兩個任務(wù)。
停止子線程
如果一切正常,，那么上面的例子很完美,。可是,，需要停止程序,，直接ctrl+c，會拋出KeyboardInterrupt錯誤,，我們修改一下主循環(huán)：
try: while True: task = rcon.rpop('queue') if not task: time.sleep(1) continue asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop) except KeyboardInterrupt as e: print(e) new_loop.stop()
可是實際上并不好使,，雖然主線程try了KeyboardInterrupt異常，但是子線程并沒有退出,，為了解決這個問題,，可以設(shè)置子線程為守護線程，這樣當(dāng)主線程結(jié)束的時候,，子線程也隨機退出,。
線程停止程序的時候，主線程退出后,，子線程也隨機退出才了,，并且停止了子線程的協(xié)程任務(wù)。
aiohttp
在消費隊列的時候,，我們使用asyncio的sleep用于模擬耗時的io操作,。以前有一個短信服務(wù)，需要在協(xié)程中請求遠程的短信api,，此時需要是需要使用aiohttp進行異步的http請求,。大致代碼如下：
server.py
/接口表示短信接口，/error表示請求/失敗之后的報警,。
async-custoimer.py
有一個問題需要注意,，我們在fetch的時候try了異常，如果沒有try這個異常,，即使發(fā)生了異常,，子線程的事件循環(huán)也不會退出。主線程也不會退出，暫時沒找到辦法可以把子線程的異常raise傳播到主線程,。（如果誰找到了比較好的方式,，希望可以帶帶我）。
對于redis的消費,，還有一個block的方法：
使用 brpop方法,，會block住task，如果主線程有消息,，才會消費,。測試了一下，似乎brpop的方式更適合這種隊列消費的模型,。
可以看到結(jié)果
協(xié)程消費
主線程用于監(jiān)聽隊列,，然后子線程的做事件循環(huán)的worker是一種方式。還有一種方式實現(xiàn)這種類似master-worker的方案,。即把監(jiān)聽隊列的無限循環(huán)邏輯一道協(xié)程中,。程序初始化就創(chuàng)建若干個協(xié)程，實現(xiàn)類似并行的效果,。
這樣做就可以多多啟動幾個worker來監(jiān)聽隊列,。一樣可以到達效果。
總結(jié)
上述簡單的介紹了asyncio的用法,，主要是理解事件循環(huán),，協(xié)程和任務(wù)，future的關(guān)系,。異步編程不同于常見的同步編程,，設(shè)計程序的執(zhí)行流的時候，需要特別的注意,。比較這和以往的編碼經(jīng)驗有點不一樣,。可是仔細想想,，我們平時處事的時候,，大腦會自然而然的實現(xiàn)異步協(xié)程。比如等待煮茶的時候,，可以多寫幾行代碼,。
相關(guān)代碼文件的Gist(https://gist.github.com/rsj217/f6b9ddadebfa4a19c14c6d32b1e961f1)
參考：Threaded Asynchronous Magic and How to Wield It
看完本文有幫助,？請轉(zhuǎn)發(fā)分享給更多人 關(guān)注「Python開發(fā)者」,，提升Python技能
|
