我第一次接觸 Python asyncio 的時候，說實話，整個人是懵的。什麼 coroutine、event loop、await——這些名詞看起來就像外星語言。但後來我發現，其實非同步程式設計的核心概念，跟我們日常生活中的行為模式一模一樣。

如果你正在學習 Python 自動化入門指南相關的技能，那 asyncio 絕對是你不能跳過的一環。這篇 Python asyncio 非同步程式設計教學，會帶你從零開始搞懂這個強大的工具。

什麼是非同步程式設計？

想像你在煮晚餐。同步的做法是：先煮飯，等飯煮好，再炒菜，等菜炒好，再煮湯。每件事都要等前一件完成才能開始。但正常人不會這樣做對吧？你會把飯放進電鍋，趁等待的時間去炒菜，菜在鍋裡翻炒時順手把湯鍋放上去。這就是非同步的精神——在等待的時候不要閒著，去做別的事。

在程式設計中，很多操作都涉及「等待」：等網路回應、等檔案讀取、等資料庫查詢。同步程式會傻傻地等，而非同步程式會在等待期間去處理其他任務，大幅提升效率。

async/await 基礎語法

Python 3.5 引入了 async 和 await 關鍵字，讓非同步程式碼寫起來幾乎跟同步一樣直覺：

import asyncio

async def fetch_data():
    print("開始抓取資料...")
    await asyncio.sleep(2)  # 模擬網路請求
    print("資料抓取完成！")
    return {"status": "ok"}

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())

async def 宣告一個協程函式（coroutine function），而 await 則是告訴 Python：「這裡需要等待，你可以先去忙別的。」這兩個關鍵字是 asyncio 的核心基礎，理解了它們，後面的概念就容易多了。

asyncio.run() 啟動非同步程式

asyncio.run() 是 Python 3.7 加入的便捷函式，它幫你處理了 event loop 的建立和關閉。在這之前，你得手動管理 event loop，程式碼會囉嘆很多。現在只要一行就搞定：

asyncio.run(main())

我個人的建議是：除非你有非常特殊的需求，否則一律用 asyncio.run() 就好。簡單、乾淨、不容易出錯。

建立與使用 Coroutine

Coroutine（協程）是 asyncio 的基本單位。用 async def 定義的函式，呼叫後不會立即執行，而是回傳一個 coroutine 物件：

async def greet(name):
    await asyncio.sleep(1)
    return f"哈囉，{name}！"

# 這樣只是建立 coroutine，不會執行
coro = greet("小明")

# 必須用 await 或丟進 event loop 才會執行
result = await coro

這個概念一開始可能有點反直覺，但習慣後你會發現它提供了極大的靈活性——你可以決定何時、以什麼順序執行這些協程。

asyncio.gather() 並行執行任務

這是 asyncio 真正展現威力的地方。asyncio.gather() 讓你同時執行多個協程：

import asyncio
import time

async def download(url, delay):
    print(f"開始下載 {url}")
    await asyncio.sleep(delay)
    print(f"下載完成 {url}")
    return f"{url} 的資料"

async def main():
    start = time.time()
    results = await asyncio.gather(
        download("api/users", 2),
        download("api/posts", 3),
        download("api/comments", 1)
    )
    elapsed = time.time() - start
    print(f"全部完成，耗時 {elapsed:.1f} 秒")
    print(results)

asyncio.run(main())

如果用同步方式，三個請求要 2+3+1=6 秒。但用 gather()，它們同時執行，只需要約 3 秒（取最慢的那個）。這在做批量 API 呼叫或爬蟲時，效能差異非常顯著。

asyncio.create_task() 任務管理

asyncio.create_task() 可以把 coroutine 包裝成 Task，讓它在背景執行：

async def background_job():
    while True:
        print("背景任務執行中...")
        await asyncio.sleep(5)

async def main():
    task = asyncio.create_task(background_job())
    # 主程式繼續做其他事
    await asyncio.sleep(12)
    task.cancel()  # 取消背景任務

跟 gather() 不同的是，create_task() 會立即開始排程執行，不需要等到 await。這在建立像 FastAPI WebSocket 教學中介紹的即時通訊應用時特別好用。

非同步程式的錯誤處理

非同步程式的錯誤處理跟同步程式差不多，用 try/except 就行。但有個常見的坑要注意：

async def risky_task():
    try:
        await asyncio.sleep(1)
        raise ValueError("出了點問題")
    except ValueError as e:
        print(f"捕獲到錯誤: {e}")
        return None

async def main():
    results = await asyncio.gather(
        risky_task(),
        risky_task(),
        return_exceptions=True  # 不讓一個錯誤中斷全部
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"任務失敗: {r}")

return_exceptions=True 是個我強烈推薦的參數。沒有它的話，任何一個任務出錯就會導致整個 gather() 拋出異常。在生產環境中，搭配 Python logging 模組教學裡介紹的日誌系統，可以更完善地追蹤這些非同步錯誤。

實戰應用範例

非同步網頁爬蟲

搭配 aiohttp 套件，可以實現高效的非同步爬蟲：

import aiohttp
import asyncio

async def fetch_page(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_page(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(f"抓取了 {len(results)} 個頁面")

asyncio.run(main())

非同步 API 呼叫

async def call_api(session, endpoint):
    async with session.get(f"https://api.example.com/{endpoint}") as resp:
        return await resp.json()

async def get_user_dashboard(user_id):
    async with aiohttp.ClientSession() as session:
        profile, posts, notifications = await asyncio.gather(
            call_api(session, f"users/{user_id}"),
            call_api(session, f"users/{user_id}/posts"),
            call_api(session, f"users/{user_id}/notifications"),
        )
        return {"profile": profile, "posts": posts, "notifications": notifications}

這個模式在實際專案中非常常見——一個頁面需要多個 API 的資料，用 asyncio 可以同時發出所有請求，大幅縮短載入時間。

asyncio vs threading vs multiprocessing

這大概是初學者最常問的問題了。我的經驗法則是：

asyncio：適合 I/O 密集型任務（網路請求、檔案讀寫、資料庫操作）。單執行緒，透過事件迴圈切換任務，記憶體開銷最小。
threading：也適合 I/O 密集型，但受限於 GIL，且執行緒切換開銷較大。適合需要與同步程式碼整合的場景。
multiprocessing：適合 CPU 密集型任務（影像處理、科學計算）。真正的平行運算，但記憶體開銷最大。

簡單來說：如果你的程式大部分時間在「等東西」，選 asyncio；如果在「算東西」，選 multiprocessing。threading 則是兩者之間的折衷方案。

總結與下一步

Python asyncio 非同步程式設計教學到這邊告一段落。回顧一下重點：用 async/await 定義協程、用 asyncio.run() 啟動、用 gather() 並行執行、用 create_task() 管理背景任務，再加上正確的錯誤處理，你就掌握了 asyncio 的核心技能。

我個人覺得 asyncio 最大的價值不是語法本身，而是它改變了你思考程式架構的方式。一旦你開始用非同步的角度思考問題，很多原本看起來很慢的程式都有了優化的空間。

想要更完整地掌握 Python 自動化技能，建議回到 Python 自動化入門指南繼續探索其他相關主題。祝你在非同步的世界裡玩得開心！

Python asyncio 非同步程式設計教學：從入門到實戰應用