前言

到目前為止，我們的程式大多是「同步」的：一步一步往下執行，上一個任務沒完成，下一個任務就得等著。這種方式雖然直覺，但在需要等待外部回應（例如網路請求、檔案存取、資料庫查詢）時，效率會變得很低。

非同步程式設計（Asynchronous Programming） 的目的，就是讓程式在等待的時候，不必閒著，而是可以去做其他事。這樣可以大幅提升整體效能，特別是在 I/O 密集的情境下。

今天會學習：

1. 非同步的核心概念（事件迴圈、協程、await、任務）。
2. 同步與非同步的效率比較，並透過生活比喻幫助理解。
3. 如何用 Python 的 async / await 撰寫非同步程式。
4. 常見的應用場景（網路爬蟲、資料庫、檔案處理）。
5. 常見錯誤與解決方式（忘記 await、使用阻塞操作、錯誤的函式庫）。

掌握非同步程式設計，能讓你寫出的程式不只是正確，更能高效處理多工需求，特別是在現代的網路應用與即時系統中。

核心概念

非同步程式設計是一種讓程式「一邊等，一邊做別的事」的模式。
它特別適合 等待外部回應 的情境，例如：

• 網路請求（下載網頁、API 呼叫）
• 檔案讀寫
• 資料庫查詢

術語與指令

生活化類比：餐廳與看診

餐廳模式

同步（傳統餐廳）

• 服務生 A：點餐 → 等廚房 → 送餐 → 收桌
• 一次只能服務一桌客人。
• 客人多時效率低。

非同步（快餐店）

• 服務生先點餐 → 交給廚房 → 再服務下一桌。
• 廚房出餐後再通知服務生。
• 整體等待時間大幅縮短。

看診模式

同步（傳統排隊）

• 醫生看診 → 病人去做檢查 → 醫生乾等。

非同步（智慧醫院）

• 醫生看診 → 病人去檢查 → 醫生先看下一位。
• 病人檢查結果出來再回到醫生。
• 整體效率提升。

同步 vs 非同步比較

適合非同步的情境

I/O 密集：網路請求、檔案、資料庫
需要等待外部服務：API、郵件發送
大量並發：聊天室、即時通訊

不適合非同步的情境

CPU 密集：AI 訓練、影像處理
簡單任務：一小段資料處理（同步更直覺）

Python 中的 async/await

基本語法範例

import asyncio

# 定義協程函式（async 表示它能被暫停）
async def fetch_data(name):
    print(f"開始處理 {name}")
    await asyncio.sleep(2)  # 模擬等待（非同步睡眠）
    print(f"{name} 處理完成")
    return f"結果: {name}"

# 主程式
async def main():
    # 建立三個任務並同時執行
    task1 = asyncio.create_task(fetch_data("任務A"))
    task2 = asyncio.create_task(fetch_data("任務B"))
    task3 = asyncio.create_task(fetch_data("任務C"))

    # 等待所有任務完成
    results = await asyncio.gather(task1, task2, task3)
    print("所有結果:", results)

# 執行事件迴圈
if __name__ == "__main__":
    asyncio.run(main())

逐行解釋

• async def：定義一個可以「暫停」的函式。
• await asyncio.sleep(2)：暫停 2 秒，這段時間讓出控制權給其他任務。
• asyncio.create_task()：把協程變成可同時執行的任務。
• asyncio.gather()：等待多個任務的結果。

執行結果比較

同步版（約 6 秒）

任務A → 任務B → 任務C

非同步版（約 2 秒）

任務A、B、C 同時開始 → 幾乎同時結束

實際應用場景

1. 網路爬蟲

import aiohttp, asyncio

async def fetch(session, url):
    async with session.get(url) as resp:
        text = await resp.text()
        return {
            "url": url,
            "status": resp.status,
            "size": len(text),
            "preview": text[:120].replace("\n", " ")
        }

async def main():
    urls = [
        "https://www.google.com"]
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*(fetch(session, u) for u in urls))
        for r in results:
            print(f"{r['status']} {r['url']} size={r['size']}")
            print(f"preview: {r['preview']}\n")

asyncio.run(main())

--執行結果如下所示，內容為網址、HTTP 狀態碼、回傳大小、前 120 字

2. 批次資料庫操作

import asyncio

async def jr_get_user(uid: int):
    # 模擬從資料庫撈資料
    await asyncio.sleep(0.1)
    return {"id": uid, "name": f"jr-user-{uid}", "email": f"jr{uid}@gmail.com"}

async def jr_update_user(user: dict):
    # 模擬更新使用者
    await asyncio.sleep(0.05)
    return {**user, "status": "updated"}

async def jr_process_users():
    ids = [1, 2, 3]
    users = await asyncio.gather(*(jr_get_user(uid) for uid in ids))
    results = await asyncio.gather(*(jr_update_user(u) for u in users))
    print(results)

asyncio.run(jr_process_users())

3. 檔案處理

import aiofiles
import asyncio

async def read_file(name):
    async with aiofiles.open(name, "r") as f:
        return await f.read()

async def main():
    files = ["test.txt", "good.txt"]
    tasks = [read_file(f) for f in files]
    contents = await asyncio.gather(*tasks)
    print(contents)

asyncio.run(main())

常見錯誤與解決方案

1. 使用阻塞操作

#  錯誤
import time
async def bad():
    time.sleep(1)  # 阻塞事件迴圈

#  正確
async def good():
    await asyncio.sleep(1)

2. 忘記加 await

#  錯誤
async def bad():
    fetch_data()  # 沒有等待

#  正確
async def good():
    result = await fetch_data()

3. 使用錯誤的函式庫

#  錯誤（阻塞）
import requests
async def bad():
    requests.get("http://example.com")

#  正確（非同步）
import aiohttp
async def good():
    async with aiohttp.ClientSession() as s:
        async with s.get("http://example.com") as resp:
            return await resp.text()

延伸閱讀

• Python 官方 asyncio 文件