【八】協程停止

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

True表示cannel成功，loop stop之後還需要再次開啟事件迴圈，最後在close，不然會報錯。

迴圈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(4)
 
    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。

import asyncio
import time
now = lambda: time.time()
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)
 

【十】新執行緒協程

新執行緒協程的話，可以在主執行緒中建立一個new_loop，然後在另外的子執行緒中開啟一個無限事件迴圈。主執行緒通過run_coroutine_threadsafe新註冊協程物件。這樣就能在子執行緒中進行事件迴圈的併發操作，同時主執行緒又不會被block。一共執行的時間大概在6s左右。

import asyncio
import time
now = lambda: time.time()
from threading import Thread
 

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)