Python 非同步程式設計(Async/Await)入門 | Python

2026/04/30

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 defawait 的基本語法、asyncio.run() 啟動協程、asyncio.gather() 並行執行任務,以及 asyncio.sleep()time.sleep() 的區別。掌握這些基礎概念,你就能在需要高效 I/O 處理的場景中,寫出更快速的 Python 程式。希望這篇文章能幫助你順利踏入非同步程式設計的世界。

如果你想複習 Python 函式的基本用法,請參考 Python 函式入門教學

關於生成器的概念與 yield 用法,請閱讀 Python 生成器教學