python 異步IO（syncio） 協程

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類的子類。保存了協程運行后的狀態，用於未來獲取協程的結果。

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)

可以看到輸出結果為：

<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

創建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對象，通過該對象可以獲取協程返回值。如果回調需要多個參數，可以通過偏函數導入。

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執行結束時候會調用回調函數。並通過參數future獲取協程執行的結果。我們創建的task和回調里的future對象，實際上是同一個對象。

future 與 result

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

可以看到輸出的結果：

Waiting:  2
Task ret:  Done after 2s
TIME:  0.0003650188446044922

阻塞和await

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

耗時的操作一般是一些IO操作，例如網絡請求，文件讀取等。我們使用asyncio.sleep函數來模擬IO操作。協程的目的也是讓這些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讓出控制權。即當遇到阻塞調用的函數的時候，使用await方法將協程的控制權讓出，以便loop調用其他的協程。現在我們的例子就用耗時的阻塞操作了。

並發和並行

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

asyncio實現並發，就需要多個協程來完成任務，每當有任務阻塞的時候就await，然后其他協程繼續工作。創建多個協程的列表，然后將這些協程注冊到事件循環中。

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)

結果如下

Waiting:  1
Waiting:  2
Waiting:  4
Task ret:  Done after 1s
Task ret:  Done after 2s
Task ret:  Done after 4s
TIME:  4.003541946411133

 

協程嵌套

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

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創建協程對象，那么await的返回值就是協程運行的結果。

 results = await asyncio.gather(*tasks)

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

不在main協程函數里處理結果，直接返回await的內容，那么最外層的run_until_complete將會返回main協程的結果。

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方式掛起協程。

 

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)

協程停止

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

• Pending
• Running
• Done
• Cancelled

創建future的時候，task為pending，事件循環調用執行的時候當然就是running，調用完畢自然就是done，如果需要停止事件循環，就需要先把task取消。可以使用asyncio.Task獲取事件循環的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)

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

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之后還需要再次開啟事件循環，最后在close，不然還會拋出異常：

Task was destroyed but it is pending!
task: <Task pending coro=<do_some_work() done,

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

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

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

start = now()

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

不同線程的事件循環

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

from threading import Thread

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

def more_work(x):
    print('More work {}'.format(x))
    time.sleep(x)
    print('Finished more work {}'.format(x))

start = now()
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

新線程協程

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

async def do_some_work(x):
    print('Waiting {}'.format(x))
    await asyncio.sleep(x)
    print('Done after {}s'.format(x))

def more_work(x):
    print('More work {}'.format(x))
    time.sleep(x)
    print('Finished more work {}'.format(x))

start = now()
new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.start()
print('TIME: {}'.format(time.time() - start))

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
(integer) 1
127.0.0.1:6379[3]> lpush queue 1
(integer) 1
127.0.0.1:6379[3]> lpush queue 1

可以看見輸出：

Waiting  2
Done 2
Waiting  5
Waiting  1
Done 1
Waiting  1
Done 1
Done 5

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

停止子線程

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

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異常，但是子線程並沒有退出，為了解決這個問題，可以設置子線程為守護線程，這樣當主線程結束的時候，子線程也隨機退出。

new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.setDaemon(True)    # 設置子線程為守護線程
t.start()

try:
    while True:
        # print('start rpop')
        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()

線程停止程序的時候，主線程退出后，子線程也隨機退出才了，並且停止了子線程的協程任務。

aiohttp

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

server.py

import time
from flask import Flask, request

app = Flask(__name__)

@app.route('/<int:x>')
def index(x):
    time.sleep(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 time
import asyncio
from threading import Thread
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()

def start_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            print(resp.status)
            return await resp.text()

async def do_some_work(x):
    print('Waiting ', x)
    try:
        ret = await fetch(url='http://127.0.0.1:5000/{}'.format(x))
        print(ret)
    except Exception as e:
        try:
            print(await fetch(url='http://127.0.0.1:5000/error'))
        except Exception as e:
            print(e)
    else:
        print('Done {}'.format(x))

new_loop = asyncio.new_event_loop()
t = Thread(target=start_loop, args=(new_loop,))
t.setDaemon(True)
t.start()

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 Exception as e:
    print('error')
    new_loop.stop()
finally:
    pass

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

try:
    while True:
        _, task = rcon.brpop("queue")
        asyncio.run_coroutine_threadsafe(do_some_work(int(task)), new_loop)
except Exception as e:
    print('error', e)
    new_loop.stop()
finally:
    pass

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

127.0.0.1:6379[3]> lpush queue 5
(integer) 1
127.0.0.1:6379[3]> lpush queue 1
(integer) 1
127.0.0.1:6379[3]> lpush queue 1

可以看到結果

Waiting  5
Waiting  1
Waiting  1
200
1 It works
Done 1
200
1 It works
Done 1
200
5 It works
Done 5

協程消費

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

import time
import asyncio
import redis

now = lambda : time.time()

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

if __name__ == '__main__':
    main()

如果使用的是 asyncio.gather創建協程對象，那么await的返回值就是協程運行的結果。

參考於：https://www.jianshu.com/p/b5e347b3a17c