Python黑魔法 --- 異步IO（ asyncio）協程

python asyncio

網絡模型有很多中，為了實作高并發也有很多方案，多線程，多程序。無論多線程和多程序，IO的排程更多取決于系統，而協程的方式，排程來自使用者，使用者可以在函數中yield一個狀态。使用協程可以實作高效的并發任務。Python的在3.4中引入了協程的概念，可是這個還是以生成器對象為基礎，3.5則确定了協程的文法。下面将簡單介紹asyncio的使用。實作協程的不僅僅是asyncio，tornado和gevent都實作了類似的功能。

event_loop 事件循環：程式開啟一個無限的循環，程式員會把一些函數注冊到事件循環上。當滿足事件發生的時候，調用相應的協程函數。

coroutine 協程：協程對象，指一個使用async關鍵字定義的函數，它的調用不會立即執行函數，而是會傳回一個協程對象。協程對象需要注冊到事件循環，由事件循環調用。

task 任務：一個協程對象就是一個原生可以挂起的函數，任務則是對協程進一步封裝，其中包含任務的各種狀态。

future：代表将來執行或沒有執行的任務的結果。它和task上沒有本質的差別

async/await 關鍵字：python3.5 用于定義協程的關鍵字，async定義一個協程，await用于挂起阻塞的異步調用接口。

上述的概念單獨拎出來都不好懂，比較他們之間是互相聯系，一起工作。下面看例子，再回溯上述概念，更利于了解。

定義一個協程

定義一個協程很簡單，使用async關鍵字，就像定義普通函數一樣：

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關鍵字定義一個協程（coroutine），協程也是一種對象。協程不能直接運作，需要把協程加入到事件循環（loop），由後者在适當的時候調用協程。asyncio.get_event_loop方法可以建立一個事件循環，然後使用run_until_complete将協程注冊到事件循環，并啟動事件循環。因為本例隻有一個協程，于是可以看見如下輸出：

Waiting: 2

TIME: 0.0004658699035644531

建立一個task

協程對象不能直接運作，在注冊事件循環的時候，其實是run_until_complete方法将協程包裝成為了一個任務（task）對象。所謂task對象是Future類的子類。儲存了協程運作後的狀态，用于未來擷取協程的結果。

task = asyncio.ensure_future(coroutine)

task = loop.create_task(coroutine)

print(task)

loop.run_until_complete(task)

可以看到輸出結果為：

<Task finished coro=<do_some_work() done, defined at /Users/ghost/Rsj217/python3.6/async/async-main.py:17> result=None>

TIME: 0.0003490447998046875

建立task後，task在加入事件循環之前是pending狀态，因為do_some_work中沒有耗時的阻塞操作，task很快就執行完畢了。後面列印的finished狀态。

asyncio.ensure_future(coroutine) 和 loop.create_task(coroutine)都可以建立一個task，run_until_complete的參數是一個futrue對象。當傳入一個協程，其内部會自動封裝成task，task是Future的子類。isinstance(task, asyncio.Future)将會輸出True。

綁定回調

綁定回調，在task執行完畢的時候可以擷取執行的結果，回調的最後一個參數是future對象，通過該對象可以擷取協程傳回值。如果回調需要多個參數，可以通過偏函數導入。

return 'Done after {}s'.format(x)

def callback(future):

print('Callback: ', future.result())

task.add_done_callback(callback)

def callback(t, future):

print('Callback:', t, future.result())

task.add_done_callback(functools.partial(callback, 2))

可以看到，coroutine執行結束時候會調用回調函數。并通過參數future擷取協程執行的結果。我們建立的task和回調裡的future對象，實際上是同一個對象。

future 與 result

回調一直是很多異步程式設計的惡夢，程式員更喜歡使用同步的編寫方式寫異步代碼，以避免回調的惡夢。回調中我們使用了future對象的result方法。前面不綁定回調的例子中，我們可以看到task有fiinished狀态。在那個時候，可以直接讀取task的result方法。

print('Waiting {}'.format(x))

print('Task ret: {}'.format(task.result()))

print('TIME: {}'.format(now() - start))

可以看到輸出的結果：

Task ret: Done after 2s

TIME: 0.0003650188446044922

阻塞和await

使用async可以定義協程對象，使用await可以針對耗時的操作進行挂起，就像生成器裡的yield一樣，函數讓出控制權。協程遇到await，事件循環将會挂起該協程，執行别的協程，直到其他的協程也挂起或者執行完畢，再進行下一個協程的執行。

耗時的操作一般是一些IO操作，例如網絡請求，檔案讀取等。我們使用asyncio.sleep函數來模拟IO操作。協程的目的也是讓這些IO操作異步化。

now = lambda: time.time()

await asyncio.sleep(x)

print('Task ret: ', task.result())

在 sleep的時候，使用await讓出控制權。即當遇到阻塞調用的函數的時候，使用await方法将協程的控制權讓出，以便loop調用其他的協程。現在我們的例子就用耗時的阻塞操作了。

并發和并行

并發和并行一直是容易混淆的概念。并發通常指有多個任務需要同時進行，并行則是同一時刻有多個任務執行。用上課來舉例就是，并發情況下是一個老師在同一時間段輔助不同的人功課。并行則是好幾個老師分别同時輔助多個學生功課。簡而言之就是一個人同時吃三個饅頭還是三個人同時分别吃一個的情況，吃一個饅頭算一個任務。

asyncio實作并發，就需要多個協程來完成任務，每當有任務阻塞的時候就await，然後其他協程繼續工作。建立多個協程的清單，然後将這些協程注冊到事件循環中。

await asyncio.sleep(x)
return 'Done after {}s'.format(x)

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.run_until_complete(asyncio.wait(tasks))

for task in tasks:

結果如下

Waiting: 1

Waiting: 4

Task ret: Done after 1s

Task ret: Done after 4s

TIME: 4.003541946411133

總時間為4s左右。4s的阻塞時間，足夠前面兩個協程執行完畢。如果是同步順序的任務，那麼至少需要7s。此時我們使用了aysncio實作了并發。asyncio.wait(tasks) 也可以使用 asyncio.gather(*tasks) ,前者接受一個task清單，後者接收一堆task。

協程嵌套

使用async可以定義協程，協程用于耗時的io操作，我們也可以封裝更多的io操作過程，這樣就實作了嵌套的協程，即一個協程中await了另外一個協程，如此連接配接起來。

await asyncio.sleep(x)
return 'Done after {}s'.format(x)

async def main():

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())

loop.run_until_complete(main())

如果使用的是 asyncio.gather建立協程對象，那麼await的傳回值就是協程運作的結果。

results = await asyncio.gather(*tasks)

for result in results:
    print('Task ret: ', result)

不在main協程函數裡處理結果，直接傳回await的内容，那麼最外層的run_until_complete将會傳回main協程的結果。

coroutine3 = do_some_work(2)

tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
]

return await asyncio.gather(*tasks)

results = loop.run_until_complete(main())

for result in results:

print('Task ret: ', result)

或者傳回使用asyncio.wait方式挂起協程。

tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
]

return await asyncio.wait(tasks)

done, pending = loop.run_until_complete(main())

for task in done:

也可以使用asyncio的as_completed方法

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))

done = loop.run_until_complete(main())

由此可見，協程的調用群組合十分靈活，尤其是對于結果的處理，如何傳回，如何挂起，需要逐漸積累經驗和前瞻的設計。

協程停止

上面見識了協程的幾種常用的用法，都是協程圍繞着事件循環進行的操作。future對象有幾個狀态：

Pending

Running

Done

Cancelled

建立future的時候，task為pending，事件循環調用執行的時候當然就是running，調用完畢自然就是done，如果需要停止事件循環，就需要先把task取消。可以使用asyncio.Task擷取事件循環的task

await asyncio.sleep(x)
return 'Done after {}s'.format(x)

try:

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()

啟動事件循環之後，馬上ctrl+c，會觸發run_until_complete的執行異常 KeyBorardInterrupt。然後通過循環asyncio.Task取消future。可以看到輸出如下：

{<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.._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.._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.._on_completion() at /Library/Frameworks/Python.framework/Versions/3.6/lib/python3.6/asyncio/tasks.py:374]>}

True

TIME: 0.8858370780944824

True表示cannel成功，loop stop之後還需要再次開啟事件循環，最後在close，不然還會抛出異常：

Task was destroyed but it is pending!

task: <Task pending coro=<do_some_work() done,

循環task，逐個cancel是一種方案，可是正如上面我們把task的清單封裝在main函數中，main函數外進行事件循環的調用。這個時候，main相當于最外出的一個task，那麼處理包裝的main函數即可。

await asyncio.sleep(x)
return 'Done after {}s'.format(x)

tasks = [
    asyncio.ensure_future(coroutine1),
    asyncio.ensure_future(coroutine2),
    asyncio.ensure_future(coroutine3)
]
done, pending = await asyncio.wait(tasks)
for task in done:
    print('Task ret: ', task.result())

task = asyncio.ensure_future(main())

print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())

不同線程的事件循環

很多時候，我們的事件循環用于注冊協程，而有的協程需要動态的添加到事件循環中。一個簡單的方式就是使用多線程。目前線程建立一個事件循環，然後在建立一個線程，在新線程中啟動事件循環。目前線程不會被block。

from threading import Thread

def start_loop(loop):

asyncio.set_event_loop(loop)

def more_work(x):

print('More work {}'.format(x))

time.sleep(x)

print('Finished more work {}'.format(x))

new_loop = asyncio.new_event_loop()

t = Thread(target=start_loop, args=(new_loop,))

t.start()

print('TIME: {}'.format(time.time() - start))

new_loop.call_soon_threadsafe(more_work, 6)

new_loop.call_soon_threadsafe(more_work, 3)

啟動上述代碼之後，目前線程不會被block，新線程中會按照順序執行call_soon_threadsafe方法注冊的more_work方法，後者因為time.sleep操作是同步阻塞的，是以運作完畢more_work需要大緻6 + 3

新線程協程

print('Done after {}s'.format(x))

asyncio.run_coroutine_threadsafe(do_some_work(6), new_loop)

asyncio.run_coroutine_threadsafe(do_some_work(4), new_loop)

上述的例子，主線程中建立一個new_loop，然後在另外的子線程中開啟一個無限事件循環。主線程通過run_coroutine_threadsafe新注冊協程對象。這樣就能在子線程中進行事件循環的并發操作，同時主線程又不會被block。一共執行的時間大概在6s左右。

master-worker主從模式

對于并發任務，通常是用生成消費模型，對隊列的處理可以使用類似master-worker的方式，master主要使用者擷取隊列的msg，worker使用者處理消息。

為了簡單起見，并且協程更适合單線程的方式，我們的主線程用來監聽隊列，子線程用于處理隊列。這裡使用redis的隊列。主線程中有一個是無限循環，使用者消費隊列。

while True:
    task = rcon.rpop("queue")
    if not task:
        time.sleep(1)
        continue
    asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)

給隊列添加一些資料：

127.0.0.1:6379[3]> lpush queue 2

(integer) 1

127.0.0.1:6379[3]> lpush queue 5

127.0.0.1:6379[3]> lpush queue 1

可以看見輸出：

Waiting 2

Done 2

Waiting 5

Waiting 1

Done 1

Done 5

我們發起了一個耗時5s的操作，然後又發起了連個1s的操作，可以看見子線程并發的執行了這幾個任務，其中5s awati的時候，相繼執行了1s的兩個任務。

停止子線程

如果一切正常，那麼上面的例子很完美。可是，需要停止程式，直接ctrl+c，會抛出KeyboardInterrupt錯誤，我們修改一下主循環：

while True:

task = rcon.rpop("queue")

if not task:

time.sleep(1)

continue

asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)

print(e)

new_loop.stop()

可是實際上并不好使，雖然主線程try了KeyboardInterrupt異常，但是子線程并沒有退出，為了解決這個問題，可以設定子線程為守護線程，這樣當主線程結束的時候，子線程也随機退出。

t.setDaemon(True) # 設定子線程為守護線程

# print('start rpop')

線程停止程式的時候，主線程退出後，子線程也随機退出才了，并且停止了子線程的協程任務。

aiohttp

在消費隊列的時候，我們使用asyncio的sleep用于模拟耗時的io操作。以前有一個短信服務，需要在協程中請求遠端的短信api，此時需要是需要使用aiohttp進行異步的http請求。大緻代碼如下：

server.py

from flask import Flask, request

app = Flask(name)

@app.route('/int:x')

def index(x):

return "{} It works".format(x)

@app.route('/error')

def error():

time.sleep(3)

return "error!"

if name == 'main':

app.run(debug=True)

/接口表示短信接口，/error表示請求/失敗之後的報警。

async-custoimer.py

import redis

import aiohttp

def get_redis():

connection_pool = redis.ConnectionPool(host='127.0.0.1', db=3)

return redis.Redis(connection_pool=connection_pool)

rcon = get_redis()

async def fetch(url):

async with aiohttp.ClientSession() as session:

async with session.get(url) as resp:

print(resp.status)

return await resp.text()

print('Waiting ', x)

ret = await fetch(url='http://127.0.0.1:5000/{}'.format(x))

print(ret)

except Exception as e:

print(await fetch(url='http://127.0.0.1:5000/error'))

else:

print('Done {}'.format(x))

t.setDaemon(True)

print('error')

pass

有一個問題需要注意，我們在fetch的時候try了異常，如果沒有try這個異常，即使發生了異常，子線程的事件循環也不會退出。主線程也不會退出，暫時沒找到辦法可以把子線程的異常raise傳播到主線程。（如果誰找到了比較好的方式，希望可以帶帶我）。

對于redis的消費，還有一個block的方法：

_, task = rcon.brpop("queue")

print('error', e)

使用 brpop方法，會block住task，如果主線程有消息，才會消費。測試了一下，似乎brpop的方式更适合這種隊列消費的模型。

可以看到結果

200

1 It works

5 It works

協程消費

主線程用于監聽隊列，然後子線程的做事件循環的worker是一種方式。還有一種方式實作這種類似master-worker的方案。即把監聽隊列的無限循環邏輯一道協程中。程式初始化就建立若幹個協程，實作類似并行的效果。

async def worker():

print('Start worker')

while True:
    start = now()
    task = rcon.rpop("queue")
    if not task:
        await asyncio.sleep(1)
        continue
    print('Wait ', int(task))
    await asyncio.sleep(int(task))
    print('Done ', task, now() - start)

def main():

asyncio.ensure_future(worker())

loop = asyncio.get_event_loop()
try:
    loop.run_forever()
except KeyboardInterrupt as e:
    print(asyncio.gather(*asyncio.Task.all_tasks()).cancel())
    loop.stop()
    loop.run_forever()
finally:
    loop.close()

main()

這樣做就可以多多啟動幾個worker來監聽隊列。一樣可以到達效果。

Python黑魔法 --- 異步IO（ asyncio）協程

繼續閱讀

來自python的【條件控制/語句循環/break/continue/else/pass】一、條件控制二、語句循環

無法解析的外部符号 wmain，該符号在函數 "void cdecl mainCRTStartupHelper(struct HINSTANCE *,unsigned short con......

TestLink導出用例轉換工具(XML2Excel)

YAML簡介和PyYAML安全操作YAML支援的類型YAML的優點：yaml的基本文法python操作

Small tricks

libsvm for python 安裝

學習軟體測試基礎測試第七天

Zeppelin 配置通路 REST APIApache Zeppelin Configuration REST API

【Torch】最簡潔logging使用指南

27. Remove Element(清單)題目代碼

Cloud Studio初體驗

使用 ctypes 進行 Python 和 C 的混合程式設計

【python】【資料處理】畫多元資料分布圖

【python】netconf協定對接管理裝置

「Python 網絡自動化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 網絡裝置

在python中建立excel并寫入

Python黑魔法 --- 異步IO（ asyncio） 協程

繼續閱讀

Python黑魔法 --- 異步IO（ asyncio）協程