Python asyncio之協程學習總結_ZenDei技術網路在線

## 實踐環境 Python 3.6.2 ## 什麼是協程 **協程**(Coroutine)一種電腦程式組件，該程式組件通過允許暫停和恢復任務，為非搶占式多任務生成子程式。**協程**也可以簡單理解為協作的程式，通過協同多任務處理實現併發的函數的變種(一種可以支持中斷的函數）。下麵，我們通過日常 ...

實踐環境

Python 3.6.2

什麼是協程

協程(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個特點：

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

有了線程為啥還要協程？

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

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

減少記憶體占用

協程的創建成本遠小於線程，可以設計得很小，小到KB級別，大大降低記憶體占用。所以，記憶體資源有限的情況下，可以創建更多協程，從而實現更高的併發。
減少上下文切換開銷，節約CPU資源

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

儘量減少可執行的線程，這樣切換次數必然會少
讓線程儘可能的處於運行狀態，而不是阻塞讓出時間片

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

其它協程示例

示例：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__()的對象，常見的可等待對象包含以下幾種：

使用async def定義的協程函數
Task對象，比如使用 asyncio.create_task() 或 asyncio.ensure_future() 創建的任務對象。
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)
```

參考連接：

https://www.shuzhiduo.com/A/gGdXlLmmd4/

作者：授客
微信/QQ：1033553122
全國軟體測試QQ交流群：7156436
Git地址：https://gitee.com/ishouke
友情提示：限於時間倉促，文中可能存在錯誤，歡迎指正、評論！
作者五行缺錢，如果覺得文章對您有幫助，請掃描下邊的二維碼打賞作者，金額隨意，您的支持將是我繼續創作的源動力，打賞後如有任何疑問，請聯繫我!!!
微信打賞支付寶打賞全國軟體測試交流QQ群

Python asyncio之協程學習總結