一、前言 #
你有沒有遇過這種場景:要處理一個幾 GB 的 log 檔,結果 Python 一口氣把整個檔案讀進記憶體,然後……記憶體爆了 💥
或者你寫了一個函式回傳一個超大的 list,但其實呼叫端只需要前 10 個元素?白白浪費了計算和記憶體。
這就是 generator(生成器) 大顯身手的時候了!Generator 讓你用「惰性運算(lazy evaluation)」的方式,一次只產生一個值,需要時才計算下一個,記憶體用量從 O(n) 降到 O(1)。
今天拍拍君就帶大家從基礎到進階,徹底搞懂 Python 的 generator 與 yield 關鍵字!
二、從 Iterator 說起 #
在深入 generator 之前,先快速回顧 Python 的 iterator protocol:
# 任何實作了 __iter__ 和 __next__ 的物件就是 iterator
class Countdown:
def __init__(self, start):
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current <= 0:
raise StopIteration
val = self.current
self.current -= 1
return val
for n in Countdown(5):
print(n) # 5, 4, 3, 2, 1
這段程式碼可以跑,但寫起來有點囉嗦——要定義一個 class、維護狀態、處理 StopIteration……
Generator 就是 Python 給你的語法糖,讓你用超簡潔的方式寫出同樣的東西 🍬
三、Generator Function 基礎 #
3.1 第一個 Generator #
只要函式裡出現 yield,它就自動變成一個 generator function:
def countdown(start):
while start > 0:
yield start
start -= 1
# 呼叫 generator function 不會執行函式內容
# 而是回傳一個 generator object
gen = countdown(5)
print(type(gen)) # <class 'generator'>
# 用 next() 逐一取值
print(next(gen)) # 5
print(next(gen)) # 4
print(next(gen)) # 3
# 當然也能用 for 迴圈
for n in countdown(3):
print(n) # 3, 2, 1
3.2 yield vs return 的差異 #
| 特性 | return |
yield |
|---|---|---|
| 函式行為 | 回傳值後結束 | 暫停並產出值,下次從暫停處繼續 |
| 呼叫結果 | 直接得到回傳值 | 得到 generator object |
| 狀態保存 | 不保存 | 保存所有區域變數 |
| 可執行次數 | 一次 | 可以多次 yield |
def simple_return():
return 1
return 2 # 永遠不會執行
def simple_yield():
yield 1
yield 2 # 第二次 next() 時會執行到這裡
yield 3 # 第三次 next() 時會執行到這裡
gen = simple_yield()
print(next(gen)) # 1
print(next(gen)) # 2
print(next(gen)) # 3
print(next(gen)) # StopIteration!
3.3 Generator 的執行流程 #
這是很多初學者容易搞混的地方,讓拍拍君用圖解說明:
def demo():
print("A: 開始執行")
yield 1
print("B: 繼續執行")
yield 2
print("C: 結束前")
gen = demo() # 什麼都不會印!只是建立 generator
val1 = next(gen) # 印出 "A: 開始執行",val1 = 1,暫停在第一個 yield
val2 = next(gen) # 印出 "B: 繼續執行",val2 = 2,暫停在第二個 yield
next(gen) # 印出 "C: 結束前",然後 raise StopIteration
重點:generator 在 yield 處暫停,保存所有狀態,下次 next() 從暫停處繼續執行。
四、Generator Expression(生成器表達式) #
就像 list comprehension 的懶惰版:
# List comprehension - 立即計算所有值,存在記憶體
squares_list = [x**2 for x in range(1_000_000)] # 占用大量記憶體
# Generator expression - 惰性計算,幾乎不占記憶體
squares_gen = (x**2 for x in range(1_000_000)) # 只是一個 generator
import sys
print(sys.getsizeof(squares_list)) # ~8,000,056 bytes (約 8 MB)
print(sys.getsizeof(squares_gen)) # 200 bytes (固定大小!)
4.1 常見用法 #
# 搭配 sum、max、min 等內建函式
total = sum(x**2 for x in range(100)) # 不需要額外括號
largest = max(len(line) for line in open("data.txt"))
# 搭配 any / all 做條件檢查
has_negative = any(x < 0 for x in numbers)
all_positive = all(x > 0 for x in numbers)
# 鏈式過濾
result = sum(
x**2
for x in range(1000)
if x % 3 == 0
if x % 5 != 0
)
4.2 什麼時候用 list comp vs generator expr? #
- 需要多次迭代 → list comprehension(generator 只能迭代一次!)
- 只需要迭代一次 → generator expression
- 資料量很大 → generator expression(省記憶體)
- 需要 indexing / slicing → list comprehension
五、實戰應用 #
5.1 處理大型檔案 #
這是 generator 最經典的應用場景:
def read_large_file(filepath):
"""逐行讀取大檔案,不需一次載入全部"""
with open(filepath, "r") as f:
for line in f:
yield line.strip()
def filter_errors(lines):
"""過濾出 ERROR 開頭的行"""
for line in lines:
if line.startswith("ERROR"):
yield line
def parse_timestamp(lines):
"""擷取時間戳記"""
for line in lines:
parts = line.split(" ", 3)
if len(parts) >= 4:
yield {"time": parts[1], "message": parts[3]}
# 管道式處理 - 超級優雅!
lines = read_large_file("server.log")
errors = filter_errors(lines)
parsed = parse_timestamp(errors)
# 整個管道在這裡才開始執行
for entry in parsed:
print(f"[{entry['time']}] {entry['message']}")
這就是所謂的 generator pipeline(生成器管道)!每一步都只在需要時才計算,記憶體用量極低。
5.2 無限序列 #
Generator 可以產生無限多個值(只要你不要全部收集起來):
def fibonacci():
"""無限費波那契數列"""
a, b = 0, 1
while True:
yield a
a, b = b, a + b
# 取前 10 個
from itertools import islice
first_10 = list(islice(fibonacci(), 10))
print(first_10) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
# 找第一個超過 1000 的費波那契數
for n in fibonacci():
if n > 1000:
print(n) # 1597
break
def infinite_id():
"""無限 ID 產生器"""
n = 1
while True:
yield f"ITEM-{n:06d}"
n += 1
id_gen = infinite_id()
print(next(id_gen)) # ITEM-000001
print(next(id_gen)) # ITEM-000002
5.3 滑動視窗(Sliding Window) #
from collections import deque
def sliding_window(iterable, size):
"""產生固定大小的滑動視窗"""
it = iter(iterable)
window = deque(maxlen=size)
# 先填滿視窗
for _ in range(size):
window.append(next(it))
yield tuple(window)
# 之後每次進一個、出一個
for item in it:
window.append(item)
yield tuple(window)
data = [1, 2, 3, 4, 5, 6, 7]
for window in sliding_window(data, 3):
print(window)
# (1, 2, 3)
# (2, 3, 4)
# (3, 4, 5)
# (4, 5, 6)
# (5, 6, 7)
六、進階:send、throw 與 close #
6.1 send() — 向 Generator 傳值 #
yield 不只能往外送值,還能接收外部傳進來的值!
def accumulator():
"""累加器:每次 send 一個數字,yield 目前的總和"""
total = 0
while True:
value = yield total # yield 同時送出 total,並接收 send 的值
if value is None:
break
total += value
gen = accumulator()
next(gen) # 啟動 generator(第一次必須用 next 或 send(None))
print(gen.send(10)) # 10 (total = 0 + 10)
print(gen.send(20)) # 30 (total = 10 + 20)
print(gen.send(5)) # 35 (total = 30 + 5)
send() 的執行流程:
- 將值傳入 generator,作為
yield表達式的結果 - Generator 從
yield處繼續執行 - 執行到下一個
yield,產出新的值
6.2 throw() — 向 Generator 拋出例外 #
def careful_generator():
try:
while True:
value = yield "等待中..."
print(f"收到: {value}")
except ValueError as e:
yield f"處理錯誤: {e}"
finally:
print("清理資源")
gen = careful_generator()
print(next(gen)) # "等待中..."
print(gen.throw(ValueError, "壞資料")) # "處理錯誤: 壞資料"
6.3 close() — 優雅關閉 Generator #
def resource_generator():
print("開啟連線")
try:
while True:
yield "資料"
except GeneratorExit:
print("收到關閉訊號,清理中...")
finally:
print("關閉連線")
gen = resource_generator()
next(gen) # "開啟連線"
gen.close() # "收到關閉訊號,清理中..." → "關閉連線"
七、yield from — 委派生成器 #
Python 3.3 引入的 yield from 讓你可以把一個 generator 委派給另一個:
# 沒有 yield from
def chain_old(*iterables):
for it in iterables:
for item in it:
yield item
# 用 yield from(更簡潔!)
def chain_new(*iterables):
for it in iterables:
yield from it
result = list(chain_new([1, 2], [3, 4], [5]))
print(result) # [1, 2, 3, 4, 5]
7.1 遞迴結構攤平 #
yield from 在處理巢狀結構時特別好用:
def flatten(nested):
"""遞迴攤平任意深度的巢狀 list"""
for item in nested:
if isinstance(item, (list, tuple)):
yield from flatten(item)
else:
yield item
data = [1, [2, [3, 4]], [5, [6, [7]]]]
print(list(flatten(data))) # [1, 2, 3, 4, 5, 6, 7]
7.2 樹的走訪 #
class TreeNode:
def __init__(self, val, left=None, right=None):
self.val = val
self.left = left
self.right = right
def inorder(node):
"""中序走訪(左 → 根 → 右)"""
if node is None:
return
yield from inorder(node.left)
yield node.val
yield from inorder(node.right)
# 4
# / \
# 2 6
# / \
# 1 3
tree = TreeNode(4,
TreeNode(2, TreeNode(1), TreeNode(3)),
TreeNode(6)
)
print(list(inorder(tree))) # [1, 2, 3, 4, 6]
八、搭配 itertools 更強大 #
Python 的 itertools 模組跟 generator 是天生一對(之前拍拍君也寫過 itertools 專文):
from itertools import chain, islice, count, takewhile, dropwhile
# chain - 串接多個 iterable
combined = chain(range(3), range(10, 13))
print(list(combined)) # [0, 1, 2, 10, 11, 12]
# islice - 切片(不需要轉成 list)
first_5_fib = list(islice(fibonacci(), 5))
# takewhile - 取到條件不成立為止
small_fibs = list(takewhile(lambda x: x < 100, fibonacci()))
print(small_fibs) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89]
# count - 無限計數器
for i, char in zip(count(1), "abc"):
print(f"{i}: {char}") # 1: a, 2: b, 3: c
九、常見陷阱與最佳實踐 #
9.1 Generator 只能迭代一次! #
gen = (x**2 for x in range(5))
print(list(gen)) # [0, 1, 4, 9, 16]
print(list(gen)) # [] ← 空的!已經耗盡了
如果需要多次迭代,用 generator function(每次呼叫產生新的 generator):
def squares(n):
return (x**2 for x in range(n))
print(list(squares(5))) # [0, 1, 4, 9, 16]
print(list(squares(5))) # [0, 1, 4, 9, 16] ← 每次都是新的!
9.2 不要在 generator 裡偷改外部狀態 #
# ❌ 不好的做法
results = []
def bad_generator(data):
for item in data:
results.append(item) # side effect!
yield item * 2
# ✅ 好的做法:保持 generator 純淨
def good_generator(data):
for item in data:
yield item * 2
9.3 注意 generator 中的例外處理 #
def safe_parse(lines):
"""安全地解析每一行,跳過格式錯誤的"""
for i, line in enumerate(lines, 1):
try:
yield int(line.strip())
except ValueError:
print(f"警告:第 {i} 行格式錯誤,已跳過")
data = ["10", "abc", "30", "", "50"]
print(list(safe_parse(data)))
# 警告:第 2 行格式錯誤,已跳過
# 警告:第 4 行格式錯誤,已跳過
# [10, 30, 50]
9.4 效能比較 #
import time
import sys
N = 10_000_000
# List approach
start = time.time()
total_list = sum([i * 2 for i in range(N)])
t_list = time.time() - start
mem_list = sys.getsizeof([i * 2 for i in range(N)])
# Generator approach
start = time.time()
total_gen = sum(i * 2 for i in range(N))
t_gen = time.time() - start
print(f"List: {t_list:.3f}s, ~{mem_list / 1024 / 1024:.0f} MB")
print(f"Generator: {t_gen:.3f}s, ~0 MB (固定)")
# 典型結果:
# List: 1.2s, ~80 MB
# Generator: 0.8s, ~0 MB (固定)
結語 #
Generator 是 Python 裡最優雅的特性之一。掌握了它,你就能:
- 🧠 省記憶體 — 處理 GB 級資料不再怕 OOM
- ⚡ 省計算 — 需要多少算多少,不做白工
- 🔗 建立 pipeline — 像水管一樣串接資料處理步驟
- ♾️ 表達無限 — 無限序列、串流資料都能優雅處理
從今天開始,看到 for 迴圈裡 append 到 list 的程式碼,先想一下:「這裡是不是可以用 yield?」
記得:能 yield 的地方就 yield,你的記憶體會感謝你的! 🎉