簡介

在之前介紹 Requests 套件時，提到系統之間經常透過網路協定中的 HTTP 進行溝通。然而，HTTP 的傳輸速度相對較慢，這是因為它基於高可靠性的 TCP 協議運行，TCP 透過三向交握和確認機制來保證數據傳輸的完整性。

以下圖為例，該圖展示了請求與回應的流程圖。首先，使用者會透過 Web Browser 向 Web Application 發送請求，接著 Web 應用程式會使用 Requests 套件向第三方平台——如 Facebook Server 發出請求，來取得使用者資訊。最終，Web Application 在收到 Facebook Server 的回應後，經過一些處理，將結果回傳給 Web 瀏覽器。由於每段傳輸可能花費數十至數百毫秒，因此整體的網路傳輸時間大約會達到數百毫秒，這在大多數情況下仍屬於使用者可接受的範圍。

然而，當 Web Application 需要與多個軟體系統互動時（如下圖所示），情況會變得更為複雜。這些請求可能包括向 Geolocation Server 請求地理位置信息，或從遠端資料庫中檢索資料。這些請求通常需要按順序執行，即每發出一個請求後，必須等待回應，再發出下一個請求。這種同步機制正是先前在介紹 Uvicorn 時提到的。因此，整體的網路等待時間會疊加，導致總延遲可能超過 2 秒，這對使用者體驗的影響非常負面。這種情況通常被稱為 IO Bound 問題，意指效能瓶頸源於輸入/輸出傳輸延遲。

本篇文章將介紹一個可以有效改善 IO Bound 問題的套件——Asyncio。它透過非同步機制來處理與 I/O 相關的任務，從而避免 Web Application 必須等待每個請求的回覆後才能繼續執行。使用 Asyncio 時，Web Application 可以同時發送多個請求，而不必按順序等待每個請求的完成。以上圖為例，當 Web Application 向 Facebook Server 發送請求時，無論 Facebook Server是否已經回覆，Web Application 都可以立刻發送下一個請求至 Geolocation Server，這大大提高了應用程式的效能和響應速度

以下圖所示，時間軸呈現了同步與非同步機制的差異。上半部分展示的是同步機制，未使用 Asyncio 套件。由於每個請求都必須等待前一個請求完成後才能執行，導致整體的等待時間較長。下半部分則是非同步機制，使用了 Asyncio 套件，因此每個請求能夠緊接著執行，大幅縮短了總體等待時間。接下來，我將通過具體範例，詳細介紹如何使用 Asyncio 來實現非同步請求。

範例

開發者無需特別安裝 Asynio 模組，因為它已經包含在 Python 標準庫中。

本篇將準備兩個範例檔案，一個展示同步機制，另一個展示非同步機制，來說明兩者的差異。由於發送請求的對象與方式並非本範例的重點，因此我們使用 sleep 來模擬回覆的等待時間，簡化範例結構。為了有效呈現同步與非同步的差異，每個回覆的等待時間將統一設為 2 秒。

首先是同步機制的範例 synchronous.py，使用 time.sleep(2) 模擬等待時間，類似於 requests 套件的行為，應用程式在等待兩秒後才會繼續執行下一步。

import time

def task(name):
    print(f"Task {name} started")
    time.sleep(2)  # This will block the execution for 2 seconds
    print(f"Task {name} finished")

def main():
    start_time = time.time()

    task('A')
    task('B')
    task('C')

    print(f"Completed in {time.time() - start_time} seconds")

if __name__ == "__main__":
    main()

執行 poetry run python synchronous.py 的結果如下：每個任務必須等待上一個任務完成後才會開始，因此整體花費約 6 秒左右。

另一個範例展示了非同步機制，該範例 asynchronous.py 使用了 Python 中的兩個保留字 await 和 async。其中，await 表示該行程式碼需要等待某個操作完成，在等待期間，應用程式可以繼續執行其他任務，從而實現非阻塞的行為。async 用來宣告一個函數為非同步任務。當函數內使用 await 關鍵字時，該函數必須以 async 進行宣告，表明其包含異步操作。在此範例使用了 await asyncio.sleep(2)，表示應用程式在等待 2 秒期間，能夠先執行其他任務。

import asyncio

async def task(name):
    print(f"Task {name} started")
    await asyncio.sleep(2)  # This won't block the event loop, allows other tasks to run
    print(f"Task {name} finished")

async def main():
    start_time = asyncio.get_event_loop().time()

    await asyncio.gather(
        task('A'),
        task('B'),
        task('C')
    )

    print(f"Completed in {asyncio.get_event_loop().time() - start_time} seconds")

if __name__ == "__main__":
    asyncio.run(main())

執行 poetry run python asynchronous.py 的結果如下：無需等待上一個任務完成，應用程式會非同步地執行下一個任務，整體只花費約 2 秒。