本文為 Python API 優化基礎 系列文，第 1 篇：
在 Python 開發中，同步（synchronous） 程式設計模式通常會讓程式執行一個任務時，必須等待該任務完成後才能繼續執行其他操作。
這樣的方式對於 I/O 密集型應用（如網路請求、檔案讀寫、資料庫查詢）來說，效能可能會受到影響，因為程式執行時可能會大量時間花在等待結果返回。
為了解決這個問題，Python 提供了 asyncio 這個標準庫，幫助開發者進行 非同步（asynchronous） 程式開發。
使程式能夠同時處理多個 I/O 任務，而不必等待其中一個任務完成後才開始下一個。這對於提升應用效能、減少等待時間有很大幫助。
本文將深入介紹 asyncio 的概念、核心功能、應用場景，並透過實際範例來展示如何在 Python 中實現非同步程式設計。
asyncio 是什麼？
asyncio 是 Python 內建的非同步編程庫，主要用於：
- 協程（coroutine）：允許函式以非同步方式運行，而不會阻塞程式。
- 事件迴圈（event loop）：管理和執行多個非同步任務，確保它們能夠有效率地運行。
- 異步 I/O（asynchronous I/O）：適用於網路請求、資料庫查詢、檔案讀寫等需要等待的操作，讓程式可以在等待時執行其他任務。
與傳統的 多執行緒（threading） 或 多行程（multiprocessing） 不同，asyncio 採用了 單執行緒+事件驅動 的方式。
透過 協程 來達成並發效果，而不需要額外創建多個執行緒或進程，因此記憶體消耗較少，適合高併發應用。
asyncio 運作機制示意圖
以下是 asyncio 的工作流程：
1️⃣ 事件迴圈（Event Loop） 啟動並監聽所有協程
2️⃣ 當一個協程遇到 I/O 操作時（如等待 API 回應），事件迴圈會讓出執行權
3️⃣ 事件迴圈在等待期間執行其他協程，不會阻塞整個程式
4️⃣ I/O 操作完成後，事件迴圈會繼續執行先前暫停的協程
這種機制能夠讓 Python 應用 同時處理多個 I/O 任務，而不需要多個執行緒或進程。
asyncio vs. 傳統同步執行的比較
| 特性 | 同步程式（Synchronous） | asyncio 非同步程式（Asynchronous） |
|---|---|---|
| 執行方式 | 順序執行，一個任務完成後才能執行下一個 | 允許多個任務同時執行（並發） |
| 適用場景 | 適合 CPU 密集型計算（數學運算、AI 訓練） | 適合 I/O 密集型應用（API、資料庫存取） |
| 效率 | 每個任務需要等待前一個完成 | 一個任務等待時，其他任務可以繼續執行 |
| 記憶體使用量 | 需要開啟多個執行緒或進程，耗費較多記憶體 | 採用單執行緒 + 協程，記憶體消耗較少 |
| 並發能力 | 需要 threading 或 multiprocessing | 透過 asyncio 管理協程，達成高併發 |
asyncio 的基本概念與關鍵詞
async / await：定義與執行協程
async 和 await 是 asyncio 的核心關鍵字，用於定義和執行非同步函式。
(1) 定義非同步函式
使用 async def 來定義一個非同步函式（協程）：
import asyncio
async def say_hello():
print("Hello")
await asyncio.sleep(2) # 模擬 I/O 操作
print("World!")
# 建立事件迴圈並執行協程
asyncio.run(say_hello())
🔹 程式解釋：
async def say_hello()定義了一個非同步函式。await asyncio.sleep(2)讓出 CPU 讓其他任務執行，而不會阻塞整個程式。asyncio.run(say_hello())啟動asyncio事件迴圈，執行協程。
執行結果（約 2 秒後輸出）：
Hello
（等待 2 秒）
World!
事件迴圈（Event Loop）
事件迴圈 是 asyncio 負責管理和執行協程的核心機制，它確保協程在適當的時機運行，而不會阻塞其他任務。
Python 3.7 以上的 asyncio.run() 會自動處理事件迴圈，但有時候我們需要手動控制事件迴圈，例如：
import asyncio
async def task1():
print("執行任務 1")
await asyncio.sleep(1)
print("任務 1 完成")
async def task2():
print("執行任務 2")
await asyncio.sleep(2)
print("任務 2 完成")
async def main():
loop = asyncio.get_running_loop() # 獲取當前事件迴圈
task1_future = loop.create_task(task1()) # 建立任務 1
task2_future = loop.create_task(task2()) # 建立任務 2
await task1_future
await task2_future
asyncio.run(main())
📌 逐行解析
async def main():
loop = asyncio.get_running_loop() # 獲取當前事件迴圈🔹 asyncio.get_running_loop() 取得當前正在運行的 事件迴圈（Event Loop）。
🔹 asyncio 的所有非同步任務都是由 事件迴圈管理，我們可以透過這個函式來操作事件迴圈，例如 手動建立非同步任務。
task1_future = loop.create_task(task1()) # 建立任務 1
task2_future = loop.create_task(task2()) # 建立任務 2
🔹 loop.create_task(task1())
task1()是一個 協程函式（coroutine function），但我們需要將它轉換為 Task 讓事件迴圈管理。loop.create_task()會立即安排task1()執行，但它不會阻塞main()，而是允許其他協程並行執行。task1_future和task2_future都是asyncio.Task物件，表示它們正在背景執行。
await task1_future
await task2_future
🔹 await task1_future
await會等待task1()完成，然後繼續執行task2()。- 但
task1()和task2()是並行運行的（因為它們是Task，並且loop.create_task()已經安排它們執行）。 await task1_future不會影響task2()的執行，它只是確保task1()完成後才繼續執行main()後續的程式碼。- 類似地，
await task2_future會確保task2()也執行完畢。
📌 這段程式碼的完整流程
1️⃣ asyncio.get_running_loop() 取得 當前事件迴圈。
2️⃣ 透過 loop.create_task(task1()) 和 loop.create_task(task2())，將 兩個任務添加到事件迴圈。
3️⃣ task1() 和 task2() 會 同時執行，因為它們是獨立的 Task。
4️⃣ await task1_future 會等待 task1() 完成，但不會影響 task2()。
5️⃣ await task2_future 會等待 task2() 完成，確保 main() 不會過早結束。
執行結果（約 2 秒完成）：
執行任務 1
執行任務 2
（1 秒後）
任務 1 完成
（再 1 秒後）
任務 2 完成
📌 create_task() vs await 的區別
如果我們 不用 create_task()，而是直接 await 兩個函式，它們會順序執行，而不是並行執行：
async def main():
await task1() # 先執行 task1()，完成後才執行 task2()
await task2() # 等 task1() 結束後才開始執行 task2()
這樣 task2() 會等 task1() 執行完後才開始，因此 無法達到並行執行的效果。
而使用 create_task()，則兩個任務可以同時運行：
async def main():
task1_future = asyncio.create_task(task1()) # 立即開始 task1
task2_future = asyncio.create_task(task2()) # 立即開始 task2
await task1_future # 確保 task1 完成
await task2_future # 確保 task2 完成
這樣 task1() 和 task2() 會同時執行，讓程式更有效率。
並行執行多個協程 (asyncio.gather)
如果我們有多個非同步函式，希望它們可以同時執行，可以使用 asyncio.gather()：
import asyncio
async def task(name, seconds):
print(f"開始 {name}")
await asyncio.sleep(seconds)
print(f"{name} 完成")
async def main():
await asyncio.gather(
task("任務 A", 2),
task("任務 B", 1),
task("任務 C", 3)
)
asyncio.run(main())這段程式碼的作用是 透過 asyncio 來並行執行多個非同步任務，避免同步執行時的等待時間。以下是程式碼的詳細解析：
📌 程式碼解析
import asyncio # 匯入 asyncio 模組，負責非同步執行
1️⃣ 定義一個非同步任務 task()
async def task(name, seconds):
print(f"開始 {name}")
await asyncio.sleep(seconds) # 模擬 I/O 等待 (不會阻塞其他協程)
print(f"{name} 完成")
🔹 說明：
- 這是一個非同步函式，使用
async def來定義。 await asyncio.sleep(seconds)模擬一個 I/O 操作（例如 API 請求、資料庫查詢等）。await代表「暫停當前任務，允許其他任務繼續執行」，不會阻塞整個應用。
2️⃣ main()：並行執行多個任務
async def main():
await asyncio.gather(
task("任務 A", 2),
task("任務 B", 1),
task("任務 C", 3)
)
🔹 說明：
asyncio.gather()可以同時執行多個協程（Coroutine），達成並發執行。await asyncio.gather(...)會等待所有任務 同時開始執行，並在 全部執行完成後 才返回結果。task("任務 A", 2)、task("任務 B", 1)、task("任務 C", 3)會同時啟動，不會彼此等待！
3️⃣ 啟動事件迴圈
asyncio.run(main())
🔹 說明：
asyncio.run(main())會自動建立事件迴圈，並執行main()協程。- Python 3.7 以上推薦用
asyncio.run()來啟動非同步程式。
📌 執行流程與結果分析
1️⃣ asyncio.gather() 啟動 task("A", 2)、task("B", 1)、task("C", 3)
2️⃣ 這三個任務同時開始，不會互相等待
3️⃣ 等到 task("B", 1) 先完成，然後 task("A", 2)，最後 task("C", 3)
執行結果：
開始 任務 A
開始 任務 B
開始 任務 C
(1 秒後)
任務 B 完成
(1 秒後)
任務 A 完成
(1 秒後)
任務 C 完成
📌 與同步執行的對比
如果改為同步執行，則所有任務都會依序等待前一個任務完成：
import time
def task(name, seconds):
print(f"開始 {name}")
time.sleep(seconds) # 阻塞程式
print(f"{name} 完成")
def main():
task("任務 A", 2)
task("任務 B", 1)
task("任務 C", 3)
main()
🔹 同步執行的結果：
開始 任務 A
(2 秒後)
任務 A 完成
開始 任務 B
(1 秒後)
任務 B 完成
開始 任務 C
(3 秒後)
任務 C 完成
⏳ 總共耗時：2 + 1 + 3 = 6 秒
asyncio 的應用場景
asyncio 適用於 I/O 密集型任務，例如：
1️⃣ 網路請求（HTTP API 呼叫）
使用 aiohttp 搭配 asyncio 來並發執行多個 API 請求。
2️⃣ 資料庫查詢
使用 aiomysql、asyncpg 等非同步資料庫庫來提高查詢效率。
3️⃣ 檔案 I/O 操作
使用 aiofiles 進行非同步讀寫，提高檔案處理速度。
4️⃣ 爬蟲與批量處理
在網路爬蟲場景中，asyncio 可讓多個爬蟲任務同時運行，提高爬取速度。
結論
asyncio 是 Python 內建的非同步程式設計庫，主要透過 協程、事件迴圈、異步 I/O 來提高應用效能，特別適用於 I/O 密集型應用。
如果你的應用場景涉及大量等待操作（如 API 請求、資料庫存取等），那麼學習 asyncio 並將其應用到專案中，將能顯著提升效能與並發能力！ 🚀