Python異步IO之協程(二):使用asyncio的不同方法實現協程

引言：在上一章中我們介紹了從yield from的來源到async的使用，並在最后以asyncio.wait()方法實現協程，下面我們通過不同控制結構來實現協程，讓我們一起來看看他們的不同作用吧～

在多個協程中的線性控制流很容易通過內置的關鍵詞await來管理。使用asyncio模塊中的方法可以實現更多復雜的結構，它可以並發地完成多個協程。

一、asyncio.wait()

你可以將一個操作分成多個部分並分開執行，而wait(tasks)可以被用於中斷任務集合(tasks)中的某個被事件循環輪詢到的任務，直到該協程的其他后台操作完成才被喚醒。

import time
import asyncio
async def taskIO_1():
    print('開始運行IO任務1...')
    await asyncio.sleep(2)  # 假設該任務耗時2s
    print('IO任務1已完成，耗時2s')
    return taskIO_1.__name__
async def taskIO_2():
    print('開始運行IO任務2...')
    await asyncio.sleep(3)  # 假設該任務耗時3s
    print('IO任務2已完成，耗時3s')
    return taskIO_2.__name__
async def main(): # 調用方
    tasks = [taskIO_1(), taskIO_2()]  # 把所有任務添加到task中
    done,pending = await asyncio.wait(tasks) # 子生成器
    for r in done: # done和pending都是一個任務，所以返回結果需要逐個調用result()
        print('協程無序返回值：'+r.result())

if __name__ == '__main__':
    start = time.time()
    loop = asyncio.get_event_loop() # 創建一個事件循環對象loop
    try:
        loop.run_until_complete(main()) # 完成事件循環，直到最后一個任務結束
    finally:
        loop.close() # 結束事件循環
    print('所有IO任務總耗時%.5f秒' % float(time.time()-start))

執行結果如下：

開始運行IO任務1...
開始運行IO任務2...
IO任務1已完成，耗時2s
IO任務2已完成，耗時3s
協程無序返回值：taskIO_2
協程無序返回值：taskIO_1
所有IO任務總耗時3.00209秒

【解釋】：wait()官方文檔用法如下：

done, pending = await asyncio.wait(aws)

此處並發運行傳入的aws(awaitable objects)，同時通過await返回一個包含(done, pending)的元組，done表示已完成的任務列表，pending表示未完成的任務列表。
注：
①只有當給wait()傳入timeout參數時才有可能產生pending列表。
②通過wait()返回的結果集是按照事件循環中的任務完成順序排列的，所以其往往和原始任務順序不同。

二、asyncio.gather()

如果你只關心協程並發運行后的結果集合，可以使用gather()，它不僅通過await返回僅一個結果集，而且這個結果集的結果順序是傳入任務的原始順序。

 

import time
import asyncio
async def taskIO_1():
    print('開始運行IO任務1...')
    await asyncio.sleep(3)  # 假設該任務耗時3s
    print('IO任務1已完成，耗時3s')
    return taskIO_1.__name__
async def taskIO_2():
    print('開始運行IO任務2...')
    await asyncio.sleep(2)  # 假設該任務耗時2s
    print('IO任務2已完成，耗時2s')
    return taskIO_2.__name__
async def main(): # 調用方
    resualts = await asyncio.gather(taskIO_1(), taskIO_2()) # 子生成器
    print(resualts)

if __name__ == '__main__':
    start = time.time()
    loop = asyncio.get_event_loop() # 創建一個事件循環對象loop
    try:
        loop.run_until_complete(main()) # 完成事件循環，直到最后一個任務結束
    finally:
        loop.close() # 結束事件循環
    print('所有IO任務總耗時%.5f秒' % float(time.time()-start))

執行結果如下：

開始運行IO任務2...
開始運行IO任務1...
IO任務2已完成，耗時2s
IO任務1已完成，耗時3s
['taskIO_1', 'taskIO_2']
所有IO任務總耗時3.00184秒

【解釋】：gather()通過await直接返回一個結果集列表，我們可以清晰的從執行結果看出來，雖然任務2是先完成的，但最后返回的結果集的順序是按照初始傳入的任務順序排的。

三、asyncio.as_completed()

as_completed(tasks)是一個生成器，它管理着一個協程列表(此處是傳入的tasks)的運行。當任務集合中的某個任務率先執行完畢時，會率先通過await關鍵字返回該任務結果。可見其返回結果的順序和wait()一樣，均是按照完成任務順序排列的。

import time
import asyncio
async def taskIO_1():
    print('開始運行IO任務1...')
    await asyncio.sleep(3)  # 假設該任務耗時3s
    print('IO任務1已完成，耗時3s')
    return taskIO_1.__name__
async def taskIO_2():
    print('開始運行IO任務2...')
    await asyncio.sleep(2)  # 假設該任務耗時2s
    print('IO任務2已完成，耗時2s')
    return taskIO_2.__name__
async def main(): # 調用方
    tasks = [taskIO_1(), taskIO_2()]  # 把所有任務添加到task中
    for completed_task in asyncio.as_completed(tasks):
        resualt = await completed_task # 子生成器
        print('協程無序返回值：'+resualt)

if __name__ == '__main__':
    start = time.time()
    loop = asyncio.get_event_loop() # 創建一個事件循環對象loop
    try:
        loop.run_until_complete(main()) # 完成事件循環，直到最后一個任務結束
    finally:
        loop.close() # 結束事件循環
    print('所有IO任務總耗時%.5f秒' % float(time.time()-start))

執行結果如下：

開始運行IO任務2...
開始運行IO任務1...
IO任務2已完成，耗時2s
協程無序返回值：taskIO_2
IO任務1已完成，耗時3s
協程無序返回值：taskIO_1
所有IO任務總耗時3.00300秒

【解釋】：從上面的程序可以看出，使用as_completed(tasks)和wait(tasks)相同之處是返回結果的順序是協程的完成順序，這與gather()恰好相反。而不同之處是as_completed(tasks)可以實時返回當前完成的結果，而wait(tasks)需要等待所有協程結束后返回的done去獲得結果。

四、總結

以下aws指：awaitable objects。即可等待對象集合，如一個協程是一個可等待對象，一個裝有多個協程的列表是一個aws。

asyncio	主要傳參	返回值順序	await返回值類型	函數返回值類型
wait()	aws	協程完成順序	(done,pending) 裝有兩個任務列表元組	coroutine
as_completed()	aws	協程完成順序	原始返回值	迭代器
gather()	*aws	傳參任務順序	返回值列表	awaitable

 【參考文獻】：

[1] Composing Coroutines with Control Structures

[2] Python 3.7.2文檔.協程與任務

[3] 控制組合式 Coroutines