Python 課程 – 迭代與生成器（Iteration & Generators）

主題：協程（coroutine）基礎

---

簡介

在 Python 的生態系統中，協程是介於一般函式與完整的執行緒之間的一種輕量級「可暫停」機制。它允許我們在同一個執行緒內，同時管理多個「工作」而不必付出多執行緒的記憶體與切換成本。

對於需要 非阻塞 I/O、串流處理 或 事件驅動 的應用程式（如網路爬蟲、即時資料分析、GUI 互動），協程提供了直觀且效能佳的解決方案。即使是剛踏入 Python 的新手，只要掌握協程的基本概念，就能寫出更具可讀性與可維護性的程式碼。

本篇文章將以 Python 3.5+ 為基礎，說明協程的核心概念、常見寫法、實務陷阱與最佳實踐，並以 3–5 個實用範例 帶領讀者一步步上手。

---

核心概念

1. 協程與生成器的關係

在 Python 中，協程最早是以 生成器（generator） 的形式實現的。生成器本身可以透過 yield 暫停執行並回傳值，而協程則在此基礎上加入了 send()、throw() 等方法，使得外部可以「向」生成器傳遞資料或拋出例外，形成雙向通訊。

重點：

• 生成器只能 單向 輸出資料（yield）。
• 協程則是 雙向 的：外部可以 send 資料進去，協程再 yield 結果出來。

2. async / await：語法糖

從 Python 3.5 起，語言本身加入了 async / await 關鍵字，讓協程的寫法更貼近同步程式的閱讀習慣。

• async def 定義一個協程函式，返回 coroutine object。
• await 用於「暫停」協程，等待另一個 awaitable（如另一個協程、Future、Task）完成後再繼續。

注意：await 只能在 async 函式內使用，否則會拋出 SyntaxError。

3. 事件迴圈（Event Loop）

協程本身不會自行執行，它需要一個 事件迴圈 來驅動。Python 標準庫的 asyncio 提供了最常用的事件迴圈實作，負責排程、I/O 多路復用與錯誤處理。

• asyncio.run(main())：在 Python 3.7+ 推薦的入口函式，會自動建立、關閉事件迴圈。
• loop.create_task(coro)：將協程包裝成 Task，交給事件迴圈執行。

---

程式碼範例

以下範例從最基礎的生成器協程逐步過渡到 async/await 的完整寫法，並附上詳細註解。

範例 1：使用 yield 與 send 的生成器協程

def accumulator():
    """一個簡易的累加器協程，使用 send 接收外部傳入的數字。"""
    total = 0
    while True:
        # 暫停並等待外部傳入的值
        value = yield total          # 第一次會得到 None
        if value is None:            # 結束條件
            break
        total += value               # 累加
        # 下一次 yield 時會回傳當前 total

# 使用方式
gen = accumulator()   # 建立 generator 物件
next(gen)              # 啟動 generator，執行到第一個 yield，回傳 0
print(gen.send(10))   # 累加 10，回傳 10
print(gen.send(5))    # 累加 5，回傳 15
gen.send(None)        # 結束協程

說明：yield 讓協程暫停，send 讓我們把資料「推」回去，形成雙向溝通。

範例 2：async / await 基礎 – 非阻塞的 Sleep

import asyncio

async def hello():
    print("Hello ...")
    # await 會讓協程暫停，讓事件迴圈去執行其他任務
    await asyncio.sleep(1)          # 非阻塞的 sleep
    print("... World!")

# 直接執行
asyncio.run(hello())

重點：asyncio.sleep 不是普通的 time.sleep，它不會阻塞整個執行緒，而是讓事件迴圈在等待期間執行其他協程。

範例 3：同時執行多個協程 – gather

import asyncio
import random

async def worker(name: str):
    delay = random.uniform(0.5, 2.0)
    await asyncio.sleep(delay)
    return f"{name} 完成（耗時 {delay:.2f}s）"

async def main():
    tasks = [
        worker("Task A"),
        worker("Task B"),
        worker("Task C")
    ]
    # gather 會同時排程所有協程，等全部完成後回傳結果列表
    results = await asyncio.gather(*tasks)
    for r in results:
        print(r)

asyncio.run(main())

說明：即使每個 worker 需要不同的等待時間，asyncio.gather 仍可一次性取得全部結果，展現協程的 高併發 能力。

範例 4：協程與 I/O – 非阻塞的 HTTP 請求（使用 aiohttp）

import asyncio
import aiohttp   # pip install aiohttp

async def fetch(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()   # 取得網頁內容

async def main():
    urls = [
        "https://www.python.org",
        "https://httpbin.org/get",
        "https://api.github.com"
    ]
    tasks = [fetch(u) for u in urls]
    pages = await asyncio.gather(*tasks)
    for i, content in enumerate(pages):
        print(f"URL {i+1} 內容長度：{len(content)}")

asyncio.run(main())

實務意義：使用 aiohttp 搭配協程，可在同一執行緒內同時發出多筆 HTTP 請求，大幅提升爬蟲或 API 客戶端的效能。

範例 5：產生器協程與 async for 結合（自訂非同步迭代器）

import asyncio

class AsyncCounter:
    """一個非同步的計數器，支援 async for 迭代。"""
    def __init__(self, limit):
        self.limit = limit
        self.current = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.limit:
            raise StopAsyncIteration
        await asyncio.sleep(0.3)   # 模擬 I/O 延遲
        self.current += 1
        return self.current

async def main():
    async for number in AsyncCounter(5):
        print(f"收到數字：{number}")

asyncio.run(main())

關鍵：實作 __aiter__ 與 __anext__ 後，即可使用 async for 以自然的語法遍歷非同步資料流。

---

常見陷阱與最佳實踐

陷阱	可能的後果	建議的解法
忘記在 async 函式內使用 await	協程不會被暫停，導致阻塞或意外的同步行為	務必在每個 I/O 操作前加上 await，或使用 asyncio.create_task 交給事件迴圈
在同步函式中直接呼叫協程 (coro() 而未 await)	只會返回 coroutine object，程式不會執行	使用 asyncio.run()（一次性）或在已有事件迴圈中 await、create_task
過度使用 asyncio.run 於同一程式多次	會在每次呼叫時重新建立/關閉事件迴圈，降低效能	在長程服務（如 Web 伺服器）中，僅在入口處呼叫一次 asyncio.run，其餘使用已存在的迴圈
忘記捕獲協程內的例外	例外會在事件迴圈關閉時被忽略，難以除錯	在協程內使用 try/except，或在 await asyncio.gather(..., return_exceptions=True) 中處理
在 CPU 密集工作中直接使用協程	協程本質仍在單執行緒，CPU 任務仍會阻塞 I/O	將 CPU 密集任務移至 concurrent.futures.ThreadPoolExecutor 或 ProcessPoolExecutor，使用 loop.run_in_executor

最佳實踐

1. 保持協程簡潔：每個協程只負責一件事，讓程式易於測試與組合。
2. 使用型別註解：async def foo() -> Awaitable[int]: 能提升 IDE 輔助與可讀性。
3. 盡量避免混用同步與非同步 API：例如在協程內使用 requests（同步）會阻塞整個事件迴圈，改用 aiohttp。
4. 資源釋放：對於需要關閉的資源（如 ClientSession、資料庫連線），使用 async with 確保自動釋放。
5. 測試：利用 pytest-asyncio 撰寫非同步測試，用 await 驗證行為。

---

實際應用場景

場景	為何適合使用協程	範例
高併發的網路爬蟲	大量 I/O（HTTP、DNS）且每筆請求時間不長	使用 aiohttp + asyncio.gather 同時爬取上千頁
即時資料流處理（WebSocket、Kafka）	需要持續接收、處理訊息且不阻塞 UI/服務端	websockets 套件搭配 async for 讀取訊息
GUI 應用程式（如 PyQt、Tkinter）	介面必須保持回應，同時執行背景任務	asyncqt 或 qasync 把事件迴圈嵌入 Qt 主迴圈
微服務與 API Server（FastAPI、Starlette）	每個請求皆為 I/O 密集，協程能提升每秒請求數	FastAPI 內建 async def 處理路由
資料庫非同步存取（PostgreSQL、MongoDB）	大量查詢/寫入操作，避免阻塞	asyncpg、motor（Mongo）提供 async 介面

---

總結

協程是 Python 針對 高併發 I/O 所提供的輕量級解決方案。從最初的 生成器協程（yield/send）到現在的 async / await 語法糖，Python 已經讓非同步程式設計變得直觀且易於維護。

• 了解 雙向通訊（send）與 事件迴圈（asyncio）的關係，是寫好協程的基礎。
• 使用 async / await 能讓程式的控制流程看起來像同步程式，降低學習門檻。
• 注意 資源釋放、例外處理 以及 避免阻塞同步 API，才能發揮協程的最大效能。

掌握了本文的概念與範例後，你就可以在 爬蟲、即時服務、GUI、微服務 等多種情境中，利用協程寫出 高效、易讀、可擴充 的 Python 程式。祝你在 Python 的非同步世界裡玩得開心！