Python 非同步程式設計(Async/Await)入門 | Python
Python 的非同步程式設計(Asynchronous Programming)讓你的程式能夠在等待 I/O 操作時繼續執行其他任務,大幅提升效率。透過 async/await 語法與 asyncio 模組,你可以輕鬆撰寫非阻塞式的並行程式。
同步 vs 非同步
在了解非同步之前,先來看看同步(Synchronous)程式的問題。同步程式會依序執行每個任務,前一個完成後才會開始下一個:
import time
def fetch_data(name, seconds):
print(f"開始取得 {name}...")
time.sleep(seconds) # 模擬耗時的 I/O 操作
print(f"{name} 完成!")
# 同步執行:總共需要 5 秒
start = time.time()
fetch_data("資料A", 2)
fetch_data("資料B", 3)
print(f"總耗時:{time.time() - start:.1f} 秒")
# 輸出:總耗時:5.0 秒
如果三個任務各需等待 2 秒,同步執行需要 6 秒,但非同步執行只需約 2 秒,因為它們可以同時等待。
async def 與 await
Python 使用 async def 定義協程(Coroutine),用 await 暫停執行並等待結果。協程是一種特殊的函式,可以在等待 I/O 時讓出控制權:
import asyncio
async def fetch_data(name, seconds):
print(f"開始取得 {name}...")
await asyncio.sleep(seconds) # 非阻塞式等待
print(f"{name} 完成!")
return f"{name} 的結果"
async def main():
result = await fetch_data("資料A", 2)
print(result)
# 執行協程
asyncio.run(main())
# 輸出:
# 開始取得 資料A...
# 資料A 完成!
# 資料A 的結果
注意:await 只能在 async def 函式內使用,而 asyncio.run() 是啟動整個非同步程式的入口。
asyncio.gather():並行執行多個任務
asyncio.gather() 可以同時執行多個協程,所有任務會並行處理,總耗時等於最慢的那個任務:
import asyncio
import time
async def fetch_data(name, seconds):
print(f"開始取得 {name}...")
await asyncio.sleep(seconds)
print(f"{name} 完成!")
return f"{name} 的結果"
async def main():
start = time.time()
# 同時執行三個任務
results = await asyncio.gather(
fetch_data("資料A", 2),
fetch_data("資料B", 3),
fetch_data("資料C", 1),
)
print(f"總耗時:{time.time() - start:.1f} 秒")
print(f"結果:{results}")
asyncio.run(main())
# 輸出:
# 開始取得 資料A...
# 開始取得 資料B...
# 開始取得 資料C...
# 資料C 完成!
# 資料A 完成!
# 資料B 完成!
# 總耗時:3.0 秒
# 結果:['資料A 的結果', '資料B 的結果', '資料C 的結果']
三個任務同時開始,總耗時只有 3 秒(最慢任務的時間),而非 6 秒。
asyncio.sleep() vs time.sleep()
這是非同步程式設計中最容易混淆的一點。time.sleep() 會阻塞整個程式,而 asyncio.sleep() 只暫停當前協程,讓其他協程繼續執行:
import asyncio
async def task_a():
print("A 開始")
await asyncio.sleep(1) # 非阻塞:其他任務可以執行
print("A 結束")
async def task_b():
print("B 開始")
await asyncio.sleep(1) # 非阻塞:其他任務可以執行
print("B 結束")
async def main():
await asyncio.gather(task_a(), task_b())
asyncio.run(main())
# 輸出:
# A 開始
# B 開始
# A 結束
# B 結束
如果將 asyncio.sleep(1) 換成 time.sleep(1),兩個任務就會變成依序執行。
實用範例:模擬並行 API 請求
以下範例模擬同時向多個 API 發送請求的場景:
import asyncio
import random
async def fetch_api(url):
"""模擬 API 請求"""
delay = random.uniform(0.5, 2.0)
print(f"請求 {url}(預計 {delay:.1f} 秒)")
await asyncio.sleep(delay)
return {"url": url, "status": 200}
async def main():
urls = [
"https://api.example.com/users",
"https://api.example.com/posts",
"https://api.example.com/comments",
]
# 並行發送所有請求
results = await asyncio.gather(*[fetch_api(url) for url in urls])
for result in results:
print(f"{result['url']} -> 狀態 {result['status']}")
asyncio.run(main())
在實際專案中,你可以搭配 aiohttp 套件來進行真正的非同步 HTTP 請求,取代上面的模擬範例。
什麼時候該用非同步?
非同步程式設計並非萬靈丹,它最適合 I/O 密集型(I/O-bound)的任務:
| 適合非同步 | 不適合非同步 |
|---|---|
| 網路請求(HTTP、API) | 大量數學運算 |
| 資料庫查詢 | 影像處理 |
| 檔案讀寫 | 資料壓縮 |
| WebSocket 通訊 | 機器學習訓練 |
對於 CPU 密集型(CPU-bound)的任務,應該使用 multiprocessing 模組或多執行緒(Threading)來達成並行處理。
總結
非同步程式設計是 Python 中處理 I/O 密集型任務的利器。本文介紹了同步與非同步的差異、async def 與 await 的基本語法、asyncio.run() 啟動協程、asyncio.gather() 並行執行任務,以及 asyncio.sleep() 與 time.sleep() 的區別。掌握這些基礎概念,你就能在需要高效 I/O 處理的場景中,寫出更快速的 Python 程式。希望這篇文章能幫助你順利踏入非同步程式設計的世界。
如果你想複習 Python 函式的基本用法,請參考 Python 函式入門教學。
關於生成器的概念與 yield 用法,請閱讀 Python 生成器教學。