生成器與協程內部原理

Python 進階主題與實務應用

---

簡介

在日常的 Python 開發中，我們常會遇到需要 大量資料逐筆處理、非阻塞 I/O 或 多任務協調 的情境。直接使用 list、for 迴圈或是傳統的多執行緒往往會帶來記憶體浪費或是程式碼複雜度升高。
生成器 (generator) 與 協程 (coroutine) 正是為了解決這類需求而設計的語言特性：它們讓程式在 需要時才產生資料，同時保留執行狀態，達到 懶評估、低記憶體佔用 與 非阻塞 的效果。

本篇文章將從 語法層面 逐步深入到 執行期的內部機制，說明 Python 如何透過 堆疊框架 (stack frame)、狀態機 與 事件迴圈 來實作生成器與協程，並提供實務上可直接套用的範例與最佳實踐。

---

核心概念

1. 生成器的基本原理

1.1 什麼是生成器？

生成器是一種 可迭代物件，其背後是一個 執行到 yield 的函式。每次呼叫 next()，函式會從上一次暫停的地方繼續執行，直到下一個 yield 或拋出 StopIteration 為止。

重點：生成器在 第一次呼叫 時不會立即執行函式體，而是返回一個 生成器物件，此物件保存了函式的 執行環境（局部變數、指令指標等）。

1.2 內部實作：堆疊框架與狀態機

Python 會為每個生成器建立一個 獨立的堆疊框架 (frame object)。當執行到 yield 時，框架會：

1. 保存局部變數與指令指標（即「暫停點」）。
2. 返回 yield 表達式的值 給呼叫者。
3. 保留框架 供下一次 next() 或 send() 時恢復。

這樣的設計讓生成器本質上是一個 協作式的狀態機，每一次切換都是由程式碼顯式觸發，而非作業系統排程。

1.3 範例：簡易的斐波那契生成器

def fibonacci(limit: int):
    """產生前 limit 個斐波那契數字，使用 yield 實作懶評估。"""
    a, b = 0, 1
    count = 0
    while count < limit:
        yield a          # 暫停，回傳 a
        a, b = b, a + b  # 恢復時繼續執行此行
        count += 1

# 使用方式
for num in fibonacci(10):
    print(num, end=' ')   # 輸出: 0 1 1 2 3 5 8 13 21 34

說明：yield 讓 fibonacci 在每次迭代時只保留必要的狀態，避免一次性產生全部資料導致記憶體暴增。

---

2. 生成器表達式 (Generator Expression)

生成器表達式與列表推導式語法相似，只是使用圓括號 () 包住，返回的是 生成器物件 而非列表。

# 產生 0~99 中的偶數，逐一計算平方
square_evens = (x*x for x in range(100) if x % 2 == 0)

# 逐項取值
for v in square_evens:
    print(v, end=' ')   # 0 4 16 36 ...

優點：一次只產生一個元素，適合 大資料流 或 串流處理。

---

3. 協程的概念與 async/await

3.1 為什麼需要協程？

傳統的阻塞 I/O（如檔案讀寫、網路請求）會讓整個執行緒卡住，造成資源浪費。協程透過 事件迴圈，讓單一執行緒在等待 I/O 時 切換到其他任務，達到類似多執行緒的併發效果，但開銷更低。

3.2 async def 與 await

• async def 定義 協程函式，呼叫時會返回一個 協程物件 (coroutine)，不會立即執行。
• await 用於 暫停 協程，等待 可等待物件（如 awaitable、Future、另一個協程）完成。

import asyncio

async def fetch_data(delay: int):
    """模擬非阻塞的 I/O 任務，delay 秒後回傳結果。"""
    await asyncio.sleep(delay)   # 讓事件迴圈暫停此協程
    return f"資料在 {delay}s 後取得"

async def main():
    # 同時啟動兩個協程
    task1 = asyncio.create_task(fetch_data(2))
    task2 = asyncio.create_task(fetch_data(3))

    # 等待全部完成，返回結果列表
    results = await asyncio.gather(task1, task2)
    for r in results:
        print(r)

# 執行事件迴圈
asyncio.run(main())

重點：asyncio.sleep 並不會真的「睡」住執行緒，而是告訴事件迴圈「這個協程在未來 X 秒後可以繼續」，允許其他協程立即取得 CPU。

3.3 協程的內部運作：awaitable 與 Future

• 每個 await 會把 協程物件 包裝成 Task（asyncio.Task），它本質上是一個 Future，代表未來某個時間點會有結果。
• 事件迴圈維護一個 待執行的任務隊列，當 Future 完成時，事件迴圈會把相應的 回呼 (callback) 加回執行隊列，繼續執行 await 之後的程式碼。

---

4. 生成器與協程的結合：async generator

Python 3.6 之後支援 非阻塞的生成器，使用 async def 搭配 yield（即 yield 前加上 await）。

import asyncio

async def async_counter(limit: int):
    """每秒產生一次計數，使用 async generator。"""
    for i in range(limit):
        await asyncio.sleep(1)   # 非阻塞等待
        yield i

async def main():
    async for number in async_counter(5):
        print(f"收到: {number}")

asyncio.run(main())

此模式在 資料流式處理（如 WebSocket、串流 API）中特別有用，能在 等待遠端資料 時保持程式的其他部分繼續運作。

---

常見陷阱與最佳實踐

陷阱	說明	解決方式
忘記關閉生成器	生成器在拋出 GeneratorExit 前不會釋放資源。	使用 try...finally 或 with contextlib.closing(gen):
在同步函式中直接 await	會產生 SyntaxError: 'await' outside async function。	必須將呼叫包在 async def，或使用 asyncio.run 執行。
過度使用 asyncio.run	每次呼叫會建立新事件迴圈，對於長程服務不友好。	在程式入口只呼叫一次，後續使用 await 或 create_task。
忘記 await 產生的 Task	任務會在背景執行，若未捕獲例外會默默失敗。	使用 await task 或 task.add_done_callback 捕獲錯誤。
產生過大的生成器表達式	雖然是懶評估，但若內部計算過於複雜會造成 CPU 峰值。	把重度運算拆分成多個小生成器，或使用 map/filter 取代。

最佳實踐

1. 保持單一職責：生成器只負責資料產生，資源釋放交給外層管理。
2. 使用 async for：遍歷 async generator 時，async for 會自動處理 await。
3. 合理設定緩衝：對於高頻 I/O，可使用 asyncio.Queue 作為生產者/消費者的緩衝區。
4. 測試與除錯：pytest-asyncio 能讓你在單元測試中輕鬆驗證協程行為。

# 例：使用 asyncio.Queue 作為緩衝
async def producer(q: asyncio.Queue, n: int):
    for i in range(n):
        await asyncio.sleep(0.1)   # 模擬 I/O
        await q.put(i)            # 放入緩衝
    await q.put(None)             # 結束訊號

async def consumer(q: asyncio.Queue):
    while True:
        item = await q.get()
        if item is None:
            break
        print(f"消費: {item}")

async def main():
    q = asyncio.Queue(maxsize=5)   # 控制緩衝大小
    await asyncio.gather(producer(q, 20), consumer(q))

asyncio.run(main())

---

實際應用場景

場景	為何使用生成器 / 協程	範例簡述
大檔案逐行處理	生成器只在需要時讀取下一行，降低記憶體佔用。	def read_large_file(path): for line in open(path): yield line
網路爬蟲 (非阻塞)	asyncio + aiohttp 可同時發送上千個請求，提升爬取速度。	async with aiohttp.ClientSession() as s: await s.get(url)
即時資料串流	async generator 搭配 WebSocket，持續接收訊息而不阻塞。	async for msg in websocket:
資料管線 (Pipeline)	生成器鏈結 (generator pipeline) 讓每個階段只處理必要的資料。	filter_gen = (x for x in source if x%2)
背景任務排程	asyncio.create_task 可在主流程之外執行長時間任務，如郵件發送。	task = asyncio.create_task(send_email())

---

總結

生成器與協程是 Python 為 懶評估、非阻塞 I/O 與 高效併發 所提供的兩大核心工具。

• 生成器透過 yield 暫停與恢復，在記憶體與 CPU 使用上提供了極佳的彈性。
• 協程則在 async/await 語法的加持下，讓 單執行緒 能同時管理多個 I/O 密集的任務，並且支援 async generator 進一步結合資料流與非阻塞特性。

掌握它們的 內部機制（堆疊框架、事件迴圈、Future）不僅能寫出更具可讀性的程式碼，也能在實務專案中 降低資源消耗、提升效能。未來你在面對大資料、即時串流或高併發服務時，只要善用這兩項特性，就能以最少的代價達成最好的結果。祝你在 Python 的進階旅程中玩得開心、寫得順手！