Python協程（中）

協程嵌套

使用async能夠定義協程，協程用於耗時的io操做，咱們也能夠封裝更多的io操做過程，這樣就實現了嵌套的協程，即一個協程中await了另一個協程，如此鏈接起來。

import asyncio
import time

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


async def main():

    tasks = [
        asyncio.ensure_future(task(1)),
        asyncio.ensure_future(task(2)),
        asyncio.ensure_future(task(4))
    ]

    dones, pendings = await asyncio.wait(tasks)
    for i in dones:
        print('Task ret: ', i.result())


start = time.time()

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
print('Time: ', time.time() - 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 task(x):
    print('Waiting: ', x)
    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)

async def main():
    coroutine1 = task(1)
    coroutine2 = task(2)
    coroutine3 = task(2)

    tasks = [
        asyncio.ensure_future(coroutine1),
        asyncio.ensure_future(coroutine2),
        asyncio.ensure_future(coroutine3)
    ]
    return await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()
results = loop.run_until_complete(main())

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

或者返回使用asyncio.wait方式掛起協程。

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


async def main():
    coroutine1 = task(1)
    coroutine2 = task(2)
    coroutine3 = task(4)

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

    return await asyncio.wait(tasks)

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 task(x):
    print('Waiting: ', x)
    await asyncio.sleep(x)
    return 'Done after {}s'.format(x)


async def main():
    coroutine1 = task(1)
    coroutine2 = task(2)
    coroutine3 = task(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))


loop = asyncio.get_event_loop()
done = loop.run_until_complete(main())

協程中止

future對象有幾個狀態：

• Pending
• Running
• Done
• Cancelled

建立future的時候，task爲pending，事件循環調用執行的時候固然就是running，調用完畢天然就是done，若是須要中止事件循環，就須要先把task取消。可使用asyncio.Task獲取事件循環的task

import asyncio

async def task(x):
    print('Waiting: ', x)

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


tasks = [
    asyncio.ensure_future(task(1)),
    asyncio.ensure_future(task(2)),
    asyncio.ensure_future(task(3))
]

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

啓動事件循環以後，立刻ctrl+c，會觸發run_until_complete的執行異常 KeyBorardInterrupt。而後經過循環asyncio.Task取消future。能夠看到輸出以下：

Waiting:  1
Waiting:  2
Waiting:  2
{<Task pending coro=<task() 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

True表示cannel成功，loop stop以後還須要再次開啓事件循環，最後在close，否則還會拋出異常：

Task was destroyed but it is pending!
task: <Task pending coro=<task() 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的兩個任務。