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

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
總時間爲4s左右。4s的阻塞時間，足夠前面兩個協程執行完畢。若是是同步順序的任務，那麼至少須要7s。此時咱們使用了aysncio實現了併發。asyncio.wait(tasks) 也可使用 asyncio.gather(*tasks) ,前者接受一個task列表，後者接收一堆task。

協程嵌套

使用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
有一個問題須要注意，咱們在fetch的時候try了異常，若是沒有try這個異常，即便發生了異常，子線程的事件循環也不會退出。主線程也不會退出，暫時沒找到辦法能夠把子線程的異常raise傳播到主線程。（若是誰找到了比較好的方式，但願能夠帶帶我）。

對於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()
這樣作就能夠多多啓動幾個worker來監聽隊列。同樣能夠到達效果。

總結

上述簡單的介紹了asyncio的用法，主要是理解事件循環，協程和任務，future的關係。異步編程不一樣於常見的同步編程，設計程序的執行流的時候，須要特別的注意。畢竟這和以往的編碼經驗有點不同。但是仔細想一想，咱們平時處事的時候，大腦會天然而然的實現異步協程。好比等待煮茶的時候，能夠多寫幾行代碼。

相關代碼文件的Gistpython 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
總時間爲4s左右。4s的阻塞時間，足夠前面兩個協程執行完畢。若是是同步順序的任務，那麼至少須要7s。此時咱們使用了aysncio實現了併發。asyncio.wait(tasks) 也可使用 asyncio.gather(*tasks) ,前者接受一個task列表，後者接收一堆task。

協程嵌套

使用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
有一個問題須要注意，咱們在fetch的時候try了異常，若是沒有try這個異常，即便發生了異常，子線程的事件循環也不會退出。主線程也不會退出，暫時沒找到辦法能夠把子線程的異常raise傳播到主線程。（若是誰找到了比較好的方式，但願能夠帶帶我）。

對於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()
這樣作就能夠多多啓動幾個worker來監聽隊列。同樣能夠到達效果。

總結

上述簡單的介紹了asyncio的用法，主要是理解事件循環，協程和任務，future的關係。異步編程不一樣於常見的同步編程，設計程序的執行流的時候，須要特別的注意。畢竟這和以往的編碼經驗有點不同。但是仔細想一想，咱們平時處事的時候，大腦會天然而然的實現異步協程。好比等待煮茶的時候，能夠多寫幾行代碼。

相關代碼文件的Gist