Python基於asyncio庫的協程實現分析

什么是協程

協程(Coroutine)一種電腦程序組件，該程序組件通過允許暫停和恢復任務，為非搶占式多任務生成子程序。協程也可以簡單理解為協作的程序，通過協同多任務處理實現並發的函數的變種(一種可以支持中斷的函數）。

下面，我們通過日常生活場景為例，對什么是協程進行說明。

假設A某在家每天都要做3件事：洗衣服(使用洗衣機)，蒸飯(使用電飯煲)，掃地(使用掃地機器人)，這三樣電器在完成任務后都會發出不一樣響聲來告訴A某事情已經完成。

這里，暫且假設A某智商有問題，每次都是嚴格按順序做這三件事：先洗完衣服，再把飯蒸好，最后才開始掃地。

接下來，我們用一段簡單的代碼來模擬上述整個過程，並記錄整個過程的耗時，其中使用了3個簡單的普通函數，分別模擬上述3件事情，如下：

import time from datetime import datetime def do_washing(): print(datetime.now(), ':開始洗衣服') time.sleep(3) # 洗衣服 # 用程序休眠來模擬過程，且別計較時間大小 print(datetime.now(), ':通知A某衣服洗好了') def steame_rice(): print(datetime.now(), ':開始蒸飯') time.sleep(5) # 蒸飯 print(datetime.now(), ':通知A某飯蒸好了') def do_clearing(): print(datetime.now(), ':開始掃地') time.sleep(2) # 掃地 print(datetime.now(), ':通知A某地掃完了') if __name__ == '__main__': startTime = time.time() do_washing() steame_rice() do_clearing() endTime = time.time() print("掃地+蒸飯+洗衣服總耗時: ", endTime - startTime) 

程序輸出：

2023-04-09 23:33:50.001204 :開始洗衣服 2023-04-09 23:33:53.002765 :衣服洗好了 2023-04-09 23:33:53.002765 :開始蒸飯 2023-04-09 23:33:58.013337 :通知A某飯蒸好了 2023-04-09 23:33:58.013337 :通知A某開始掃地 2023-04-09 23:34:00.024784 :通知A某地掃完了 掃地+蒸飯+洗衣服總耗時: 10.023579835891724 

直到有一天，A某的朋友來他家做客，體驗到他的“高效”辦事效率后，建議他不用等每件事情都做完才做下一件事情。A某聽后，虛心采納，並告訴自己要開始培養新的習慣。

第二天開始呢，A某開始改變自己，把衣服扔洗衣機，並啟動機洗程序后，就去淘米蒸飯了，等電飯煲開始蒸飯后，就去清掃地板了。

接下來，我們對上述代碼進行稍微修改，以便模擬上述過程，並記錄整個過程的耗時，如下：

import time from datetime import datetime import asyncio async def do_washing(): print(datetime.now(),':開始洗衣服') await asyncio.sleep(3) print(datetime.now(),':通知A某衣服洗好了') async def do_clearing(): print(datetime.now(), ':開始掃地') await asyncio.sleep(5) print(datetime.now(), ':通知A某地掃完了') async def steame_rice(): print(datetime.now(), ':開始蒸飯') await asyncio.sleep(2) print(datetime.now(), ':通知A某飯蒸好了') tasks = [ do_washing(), steame_rice(), do_clearing() ] if __name__ == '__main__': loop = asyncio.get_event_loop() start_time = time.time() loop.run_until_complete(asyncio.wait(tasks)) loop.close() end_time = time.time() print("掃地+蒸飯+洗衣服總耗時: ", end_time - start_time) 

程序輸出：

2023-04-09 23:35:17.422790 :開始掃地 2023-04-09 23:35:17.422790 :開始蒸飯 2023-04-09 23:35:17.422790 :開始洗衣服 2023-04-09 23:35:19.427500 :通知A某飯蒸好了 2023-04-09 23:35:20.427813 :通知A某衣服洗好了 2023-04-09 23:35:22.429780 :通知A某地掃完了 掃地+蒸飯+洗衣服總耗時: 5.0069899559021 

不得不誇A某進步真大，相比之前，這次耗時減少了近一半。

以上這段代碼就是協程的簡單實現，充分體現了協程的3個特點：

1. 多任務並行：A某同時完成了3項任務--分別代表3個協程。
2. 異步任務：3項任務中，沒有一項是需要A某在一旁一直看着直到做完的，每項任務開啟后，A某都可以離開去做別的任務。
3. 協作式(非搶占式)：每項任務能否“占用”A某，取決於A某是否正被其它任務“占用”，即是否有任務主動“讓出”A某，不是靠“搶占”，更像是協商。

有了線程為啥還要協程？

協程是用戶視角的一種抽象，操作系統並沒有這個概念，其主要思想是在用戶態實現調度算法，用少量線程完成大量任務的調度。

相對線程而言，協程具備以下優勢：

• 減少內存占用

  協程的創建成本遠小於線程，可以設計得很小，小到KB級別，大大降低內存占用。所以，內存資源有限的情況下，可以創建更多協程，從而實現更高的並發。

• 減少上下文切換開銷，節約CPU資源

如上圖，線程之間的切換請求，由系統內核來實現，而協程之間的切換，則可由用戶自由控制，即交由用戶態的代碼來完成，極大程度避免了系統內核級線程上下文切換造成的CPU資源浪費。具體實現思路如下：

1. 盡量減少可執行的線程，這樣切換次數必然會少

2. 讓線程盡可能的處於運行狀態，而不是阻塞讓出時間片

一個線程可以擁有多個協程，主要注意的是，一個線程內的多個協程卻是串行的，無論CPU有多少個核，因為協程本質上還是一個函數，當一個協程運行時，其它協程必須掛起。實際開發過程中，可以使用協程在將一些耗時的IO操作異步化，例如寫文件、耗時IO請求等來提升程序執行效率。

相關語法說明

接下來，就上面的例子，對協程相關語法進行說明。

async def do_washing() 

使用async def語法定義協程函數do_washing。

協程函數示例：

async def func(param1, param2): do_stuff() await some_coroutine() 

注意：

1. 使用async def語法定義的函數始終是協程函數，即使它們不包含wait或async關鍵字。

2. 采用傳統的函數調用方式，直接調用協程函數，函數不會被立即執行，會產生類似RuntimeWarning: coroutine 'xxxx協程函數' was never awaited的告警日志，並返回一個協程對象。僅運行事件循環時才會運行協程。

3. await 掛起當前協程以等待一個可等待(awaitable)對象--協程函數或者實現了__await__()的對象，直到可等待對象返回結果。可以將這個可等待對象，簡單的理解為待執行的異步任務(一般是比較耗時的任務，比如開篇示例中用作比擬的煲飯）。

   注意：

   1. await只能在協程函數內部使用。

   2. 程序遇到await關鍵詞時，會將程序控制權交給主程序，由主程序分配給其它協程。當可等待對象返回結果，並且此時程序控制權還被其它協程占用時，則被掛起的協程依舊無法繼續往下運行，直到獲取程序控制權。關於這個結論，可用下述示例代碼進行驗證：

      from datetime import datetime import asyncio async def do_washing(): print(datetime.now(),':開始洗衣服') await asyncio.sleep(0.5) for i in range(10000): if i % 4000 == 0: print('洗衣服') print(datetime.now(),':衣服洗好了') async def do_cooking(): print(datetime.now(), ':開始煲飯') for i in range(100000): if i%20000 == 0: print('煲飯') await asyncio.sleep(5) print(datetime.now(), ':飯煲好了') tasks = [ do_cooking(), do_washing() ] if __name__ == '__main__': loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) loop.close() 

      輸出：

      2023-04-10 23:53:37.804727 :開始洗衣服 2023-04-10 23:53:37.804727 :開始煲飯 煲飯 煲飯 煲飯 煲飯 煲飯 洗衣服 洗衣服 洗衣服 2023-04-10 23:53:38.310586 :衣服洗好了 2023-04-10 23:53:42.811876 :飯煲好了 

asyncio.sleep(2) 

給定秒數后完成的協程--阻塞指定的秒數。sleep函數還可以指定result參數，協程完成時將該參數值返回給調用者（默認返回None），如下：

result = await asyncio.sleep(0.5, result='task done') print(result) # 輸出：task done 

sleep總是會掛起當前任務，以允許其他任務運行。可以利用這個特性，將秒數設置為0，即asyncio.sleep(0)，以便提供一個經優化的路徑以允許其他任務運行。 這可供長時間運行的函數使用，避免調用該函數時阻塞事件循環。

asyncio.get_event_loop()

為當前上下文獲取事件循環(event loop)，返回一個實現了AbstractEventLoop接口的事件循環對象。如果沒有為當前上下文設置任何事件循環，且當前策略沒有指定創建一個事件循環，則拋出異常。必須返回非None值。

AbstractEventLoop.run_until_complete(asyncio.wait(tasks))

運行直到asyncio.wait(tasks)運行完成。返回asyncio.wait(tasks)的運行結果，或者拋出異常。

asyncio.run(coro, *, debug=False) 

執行協程 coro 並返回結果。

此函數會運行傳入的協程，負責管理 asyncio 事件循環，終結異步生成器，並關閉線程池。

當有其他 asyncio 事件循環在同一線程中運行時，此函數不能被調用。

如果debug 為 True，事件循環將以調試模式運行。

此函數總是會創建一個新的事件循環並在結束時關閉之。它應當被用作 asyncio 程序的主入口點，理想情況下應當只被調用一次。

示例:

async def main(): await asyncio.sleep(1) print('hello') asyncio.run(main()) 

3.7 新版功能.

asyncio.wait(tasks) 

具備完整參數列表的wait函數定義如下

asyncio.wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED) 

並發地運行 fs可迭代對象中的可等待對象，並進入阻塞狀態直到滿足return_when參數所指定的條件(缺省參值為ALL_COMPLETED)。

注意，aws參數不能為空。

函數返回 Future 集合: (done, pending)。

請注意，此函數不會引發 asyncio.TimeoutError。當超時發生時，未完成的 Future 將在指定秒數后被返回。

return_when 指定此函數應在何時返回，可選值如下：

• FIRST_COMPLETED

  函數將在任意可等待對象結束或取消時返回。

• FIRST_EXCEPTION

  函數將在任意可等待對象因引發異常而結束時返回。當沒有引發任何異常時它就相當於ALL_COMPLETED。

• ALL_COMPLETED

  函數將在所有可等待對象結束或取消時返回。

其它協程示例

示例：Hello world攜程

import asyncio async def hello_world(): print("Hello World!") return 'hello world' # print(hello_world()) # RuntimeWarning: coroutine 'hello_world' was never awaited #<coroutine object compute at 0x000001B6265F08E0> loop = asyncio.get_event_loop() # Blocking call which returns when the hello_world() coroutine is done res = loop.run_until_complete(hello_world()) # 把協程對象傳遞給事件循環 print(res) # 輸出：hello world loop.close() 

python3.7版本，也可以使用新API asyncio.run來簡化代碼

import asyncio async def hello_world(): print("Hello World!") return 'hello world' asyncio.run(hello_world()) 

示例：顯示當前日期

使用sleep()函數在5秒內每1秒顯示一次當前日期的協程示例

import asyncio import datetime async def display_date(loop): end_time = loop.time() + 5.0 while True: print(datetime.datetime.now()) if (loop.time() + 1.0) >= end_time: break await asyncio.sleep(1) loop = asyncio.get_event_loop() # Blocking call which returns when the display_date() coroutine is done loop.run_until_complete(display_date(loop)) loop.close() 

示例: 鏈式協程(Chain coroutines)

import asyncio async def compute(x, y): print("Compute %s + %s ..." % (x, y)) await asyncio.sleep(1.0) return x + y async def print_sum(x, y): result = await compute(x, y) print("%s + %s = %s" % (x, y, result)) loop = asyncio.get_event_loop() loop.run_until_complete(print_sum(1, 2)) loop.close() 

compute()被鏈接到print_sum()：print_sum()協程等待compute()完成后再返回結果

示例的序列圖

“Task”是由AbstractEventLoop.run_until_complete()方法在獲取協程對象而不是任務時創建的。

該圖顯示了控制流程，但並沒有確切描述事物內部是如何工作的。例如，sleep協程創建了一個內部future，它使用AbstractEventLoop.call_later()在1秒內喚醒任務。

可等待對象

整體而言，python協程的可等待對象包含協程函數或者實現了__await__()的對象，常見的可等待對象包含以下幾種：

1. 使用async def定義的協程函數

2. Task對象，比如使用 asyncio.create_task() 或 asyncio.ensure_future() 創建的任務對象。

3. Future對象，比如使用 asyncio.Future() 創建的對象。

Future

Future，是對協程的封裝，代表一個異步操作的最終結果--將來執行或沒有執行的任務的結果，其值會在將來被計算出來。

class asyncio.Future(*, loop=None)

該類基本兼容concurrent.futures.Future。

差別：

• result()和exception()不接受超時參數，並且在future尚未完成時引發異常。
• 總是通過事件循環的call_soon_threadsafe()調用使用add_done_callback()注冊的回調。
• 該類與concurrent.futures包中的wait()和as_completed()函數不兼容。

該類不是線程安全的。

類方法

• cancel() 
  取消future並安排執行回調

  如果future已經完成或者取消，則返回False。否則，修改future的狀態為已取消，並安排執行回調，並返回True。

• cancelled()

  如果future已取消則返回True。

• done() 
  如果future已完成則返回True。

  已完成意味着可獲取結果或者異常，或者future已被取消。

• result()

  返回future呈現的結果。

  如果future已被取消，則引發CancelledError。如果future的結果還不可獲取，則會引發InvalidStateError。如果future已完成並且存在異常，則該異常會被拋出。

• exception()
  返回給future設置的異常。

  只有在future完成時，才會返回異常（如果未設置異常，則返回None）。如果future已被取消，則引發CancelledError。如果future尚未完成，則會引發InvalidStateError。

• add_done_callback(fn)

  添加一個回調，以便在future完成時運行。

  使用一個future對象作為參數調用回調。如果調用時，future已經完成，則使用call_soon()調用回調。

  使用functools.partial將參數傳遞給回調。例如

  fut.add_done_callback(functools.partial(print, "Future:", flush=True)) 將調用print("Future:", fut, flush=True)

• remove_done_callback(fn)

​ 從“call when done”列表中刪除回調的所有實例。

​ 返回已刪除的回調數。

• set_result(result)
  標記future為已完成並設置其結果。

  如果調用此方法時future已完成，則會引發InvalidStateError

• ​ set_exception(exception)

  標記future為已完成並設置一個異常。

  如果調用此方法時future已完成，則會引發InvalidStateError。

例子： Future配合run_until_complete()的使用

import asyncio async def slow_operation(future): await asyncio.sleep(1) future.set_result('Future is done!') loop = asyncio.get_event_loop() future = asyncio.Future() asyncio.ensure_future(slow_operation(future)) loop.run_until_complete(future) print(future.result()) # Future is done! loop.close() 

協程函數負責計算(耗時1秒)，並將結果存儲到future。run_until_complete()方法等待future的完成。

注意：
run_until_complete() 方法在內部使用add_done_callback()方法，以便在future完成時得到通知。

Future類封裝了可調用對象的異步執行

示例：Future配合run_forever()的使用

可以使用Future.add_done_callback()方法以不同的方式編寫前面的示例，以明確描述控制流：

import asyncio async def slow_operation(future): await asyncio.sleep(1) future.set_result('Future is done!') def got_result(future): print(future.result()) loop.stop() loop = asyncio.get_event_loop() future = asyncio.Future() asyncio.ensure_future(slow_operation(future)) future.add_done_callback(got_result) try: loop.run_forever() finally: loop.close() 

在本例中，future用於將slow_operation()鏈接到got_result()：當slow_ooperation()完成時，將調用got_resull()獲取結果

Task

class asyncio.Task(coro, *, loop=None)
安排協程的執行：將其封裝在future。Task是Future的一個子類。

task負責在事件循環中執行協程。如果封裝的協程由future生成，則task將阻塞執行封裝的協程並等待future的完成。當future完成並返回結果或者異常，封裝的協程的執行將重新開始，並檢索future的結果或異常。

事件循環使用協作調度：一個事件循環一次只運行一個task。如果其他事件循環在不同的線程中運行，則其他task可以並行運行。當task等待future完成時，事件循環會執行一個新task。

取消一項task和取消一個future是不同的。調用cancel()將向封裝的協程拋出CancelledError。僅當封裝的協程沒有捕獲CancelledError異常或拋出CancelledError異常時，cancelled()才會返回True。

如果一個掛起的task被銷毀，則其封裝的協程不會被執行完。這可能是一個bug，並記錄一條警告：

Task was destroyed but it is pending! task: <Task pending coro=<kill_me() done, defined at test.py:5> wait_for=<Future pending cb=[Task._wakeup()]>> 

不要直接創建Task實例：使用ensure_future()函數或AbstractEventLoop.create_task()方法。

這個類不是線程安全的。

類方法

• all_tasks(loop=None)
  返回給定事件循環的所有任務集。默認返回當前事件循環的所有任務。

• current_task(loop=None) 
  返回給定事件循環中當前正在運行的任務。默認返回當前事件循環中的當前任務。

  不在Task上下文中調用該函數時返回None

• cancel()
  請求取消任務

  安排在事件循環的下一個循環中將CancelledError拋出到封裝的協程中。然后，協程有機會使用try/except/finally清理甚至拒絕請求。

  與Future.cancel()不同，這並不能保證task會被取消：異常可能會被捕獲並采取行動，從而延遲task的取消或完全阻止取消。該task也可能返回一個值或拋出一個不同的異常。

  調用此方法后，cancelled()將不會立即返回True（除非任務已被取消）。當封裝的協程以CancelledError異常終止時，task將被標記為已取消（即使未調用cancel()）。

• get_stack(*, limit=None)

  返回此任務的協程的堆棧幀列表。

  如果協程沒有完成，則返回它被掛起的堆棧。如果協同程序已成功完成或被取消，則返回一個空列表。如果協同程序被異常終止，則返回traceback幀列表。

  堆棧幀總是按從舊到新的順序排列。

  可選limit給出了要返回的最大幀數；默認情況下，將返回所有可獲取的幀。它的含義因返回堆棧還是trackback而不同：返回堆棧的最新幀，但返回traceback的最舊幀（這與traceback模塊的行為相符）。

  由於我們無法控制的原因，對於掛起的協程，只返回一個堆棧幀。

• print_stack(*, limit=None, file=None)

  打印此任務的協程的堆棧或traceback。

  為get_stack()檢索的幀生成類似於traceback模塊的輸出。limit參數被傳遞給get_stack()。file參數為I/O流，輸出將寫入該流；默認情況下，輸出寫入sys.stderr

示例：並行執行task

並行執行3個task (A, B, C)

import asyncio async def factorial(name, number): f = 1 for i in range(2, number+1): print("Task %s: Compute factorial(%s)..." % (name, i)) await asyncio.sleep(1) f *= i print("Task %s: factorial(%s) = %s" % (name, number, f)) loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.gather( factorial("A", 2), factorial("B", 3), factorial("C", 4), )) loop.close() 

輸出：

Task B: Compute factorial(2)... Task C: Compute factorial(2)... Task A: Compute factorial(2)... Task B: Compute factorial(3)... Task C: Compute factorial(3)... Task A: factorial(2) = 2 Task B: factorial(3) = 6 Task C: Compute factorial(4)... Task C: factorial(4) = 24 

task在創建時會自動被安排執行。事件循環將在所有task完成后停止。

Task函數

注意：

在下面的函數中，可選的循環參數允許顯式設置底層task或協程使用的事件循環對象。如果沒有提供，則使用默認的事件循環

• asyncio.as_completed(fs, *, loop=None, timeout=None)

  返回一個迭代器，該迭代器在等待時為Future實例。

  如果在所有Future完成之前發生超時，則引發asyncio.TimeoutError。

  示例：

  for f in as_completed(fs): result = yield from f # The 'yield from' may raise # Use result 

  注意：
  future f不一定是fs的成員

• asyncio.ensure_future(coro_or_future, *, loop=None)

  安排協程對象的執行：在其封裝在Future中。返回一個Task對象。

  如果參數是Future，則直接返回。

  版本3.4.4中新增

  版本3.5.1變更: 函數接受任何可等待對象。

• asyncio.async(coro_or_future, *, loop=None) 
  廢棄的ensure_future()的別名

  版本 3.4.4開始廢棄

• asyncio.wrap_future(future, *, loop=None)

  將concurrent.futures.Future對象封裝在Future對象中。

• asyncio.gather(*coros_or_futures, loop=None, return_exceptions=False)

  返回來自給定協程對象或future的future聚合結果。

  所有future必須共享相同的事件循環。如果所有task都成功完成，那么返回的future結果就是結果列表(按照原始序列的順序，不一定是結果到達的順序）。如果return_exceptions為true，則task中的異常將被視為成功的結果，並收集在結果列表中；否則，第一個拋出的異常將立即傳遞給返回的future。

  取消：如果外部Future被取消，則所有子項(尚未完成)也將被取消。如果任何子項被取消，這將被視為引發CancelledError錯誤——在這種情況下，外部Future不會被取消。（這是為了防止取消一個子項而導致其他子項被取消。）

• asyncio.iscoroutine(obj)
  如果obj是一個協程對象，該對象可能基於生成器或async def協程，則返回True。

• asyncio.iscoroutinefunction(func) 
  如果func被判斷為協程函數，則返回True，協程函數可以是被修飾的生成器函數或async def函數。

• asyncio.run_coroutine_threadsafe(coro, loop)

  向給定的事件循環提交一個協程對象。

  返回concurrent.futures.Future以訪問結果。

  該函數被從不同於運行事件循環線程的線程調用。用法：

  # Create a coroutine coro = asyncio.sleep(1, result=3) # Submit the coroutine to a given loop future = asyncio.run_coroutine_threadsafe(coro, loop) # Wait for the result with an optional timeout argument assert future.result(timeout) == 3 

  如果在協程中引發異常，則會通知返回的future。它還可以用於取消事件循環中的task：

  try: result = future.result(timeout) except asyncio.TimeoutError: print('The coroutine took too long, cancelling the task...') future.cancel() except Exception as exc: print('The coroutine raised an exception: {!r}'.format(exc)) else: print('The coroutine returned: {!r}'.format(result)) 

  注意：
  與模塊中的其他函數不同，run_coroutine_threadsafe() 要求顯式傳遞loop參數。

  版本3.5.1中新增

• coroutine asyncio.sleep(delay, result=None, *, loop=None)

  創建一個給定秒數后完成的協程--阻塞指定的秒數。sleep函數還可以指定result參數，協程完成時將該參數值返回給調用者（默認返回None）

• asyncio.shield(arg, *, loop=None)
  等待future，保護它不被取消。

  語句:

  res = yield from shield(something()) 

  等價於:

  res = yield from something() 

  除非包含它的協程被取消，否則在something()中運行的任務不會被取消。從something()的視角來看，並沒法生取消。但是它的調用者仍然被取消，所以yield from表達式仍然會引發CancelledError。注意：如果通過其他方式取消了something()，這仍然會取消shield()。

  如果你想完全忽略取消（cancellation,不推薦），你可以將shield()與try/except子句結合使用，如下所示:

  try: res = yield from shield(something()) except CancelledError: res = None 

• coroutine asyncio.wait(futures, *, loop=None, timeout=None, return_when=ALL_COMPLETED)

  等待futures序列參數給定的Future和協程對象執行完成。協程將被封裝在task中。返回兩個Future集：(done，pending)。

  • futures序列參數不能為空。

  • timeout參數可用於控制返回前等待的最大秒數。timeout可以是int或float類型。如果未指定timeout參數或參數值為空，則沒有等待時間限制，即永不超時。

  • return_when指示此函數何時返回。它必須是concurrent.futures模塊的以下常量之一：

    • FIRST_COMPLETED 當任何future完成或被取消時，函數將返回。
    • FIRST_EXCEPTION 當任何future因為引發異常而結束時，函數將返回。如果沒有future引發異常，那么它相當於ALL_COMPLETED。
    • ALL_COMPLETED當所有future結束或被取消時，函數將返回。

  這個函數是一個協程。

  用法:

  done, pending = yield from asyncio.wait(fs) 

  注意
  這不會引發asyncio.TimeoutError。pending集合中存放的是發生超時時未完成的future。

• coroutine asyncio.wait_for(fut, timeout, *, loop=None)

  等待單個future或協程對象完成直到發生超時(如果超時限制的話)。如果timeout為None，則一直等待直到future完成。

  協程將被封裝在Task中。

  函數返回Future或協同程序的結果。當發生超時時，將取消task並拋出asyncio.TimeoutError。為了避免任務取消，請將其封裝在shield()中。

  如果取消wait，那么future fut也將被取消。

  該函數為一個協程，用法：

  result = yield from asyncio.wait_for(fut, 60.0)