進階前九課已經涵蓋物件導向、函數式寫法、並行、資料處理、Web、資料庫、測試與部署。本課補語言本身還缺獨立教材的兩個協定——迭代/產生器協定與情境管理器協定——並把第 3 課 03_concurrency_parallelism.md 示範過的三種並行,用 GIL 的角度收攏成一張選擇表。範例以 CPython 的預設組建為準。
相關教材:02_intermediate_level/01_advanced_functions.md(裝飾器、閉包)、02_intermediate_level/02_file_handling.md(with open)、03_concurrency_parallelism.md(threading / multiprocessing / asyncio 的完整範例)、02_functional_programming.md(itertools、functools)。
目錄
1. 迭代器協定
for 迴圈只依賴兩個方法:
__iter__()回傳一個迭代器;- 迭代器的
__next__()回傳下一個值,沒有值時丟出StopIteration。
class Countdown:
def __init__(self, start):
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current <= 0:
raise StopIteration
self.current -= 1
return self.current + 1
print(list(Countdown(3))) # [3, 2, 1]
串列、字典、集合、字串、檔案物件都是可迭代物(iterable):呼叫 iter(x) 會得到一個新的迭代器。迭代器本身也是可迭代物(__iter__ 回傳自己),這就是為什麼 for 可以直接吃迭代器。
| 物件 | 可迭代 | 是迭代器 | 可以走幾次 |
|---|---|---|---|
[1, 2, 3] |
是 | 否 | 無限次(每次 iter() 都是新的) |
iter([1, 2, 3]) |
是 | 是 | 一次 |
range(10**9) |
是 | 否 | 無限次,而且不佔記憶體 |
檔案物件 open(p) |
是 | 是 | 一次(除非 seek(0)) |
2. 產生器函式與 yield
函式體裡只要出現 yield,這個函式就變成產生器函式:呼叫它不會執行內容,而是回傳一個產生器物件,它同時實作了 __iter__ 與 __next__。每次 __next__ 執行到下一個 yield,把值交出去,所有區域變數留在原地。
def count_up(n):
i = 1
while i <= n:
yield i
i += 1
g = count_up(3)
print(next(g), next(g), next(g)) # 1 2 3
# next(g) 再呼叫一次會丟出 StopIteration
count_up(10**9) 不會配置十億個整數,這是它和 list(range(10**9)) 的差別。
三個常被忽略的細節:
- 產生器只能走一次。 走完再迭代,不會重新執行函式,而是什麼都沒有。需要第二次,就再呼叫一次函式。
return value會變成StopIteration.value。 用yield from委派時,子產生器的回傳值就是yield from運算式的值。yield from other把另一個可迭代物的值逐一交出去,用來串接,而不是先做成串列。
產生器運算式用圓括號:(x*x for x in nums) 是延遲的;[x*x for x in nums] 是串列,會立刻算完。兩個括號不要看走眼。
產生器也可以接收值
value = yield x 讓呼叫端用 g.send(v) 把資料送回產生器內部。這是 asyncio 早期協程的基礎,現在直接寫 async def 即可,但讀舊程式碼時會遇到。
3. 產生器管線與 itertools
把每個步驟寫成一個產生器,再串起來,就得到一條只在需要時才流動的管線:
import itertools
def read_lines(path):
with open(path, encoding="utf-8") as f:
for line in f:
yield line.rstrip("\n")
def non_empty(lines):
return (ln for ln in lines if ln.strip())
def numbered(lines):
for i, ln in enumerate(lines, 1):
yield f"{i:>4}: {ln}"
first_ten = itertools.islice(numbered(non_empty(read_lines("big.log"))), 10)
print("\n".join(first_ten))
整個過程只讀到第 10 個非空行為止,不會把 big.log 載進記憶體。
itertools 裡最常用的幾個:
| 函式 | 用途 |
|---|---|
islice(it, n) / islice(it, a, b) |
切片,不需要先轉成串列 |
chain(a, b) |
把多個可迭代物接成一條 |
groupby(sorted_it, key) |
相鄰相同鍵分組(要先排序) |
accumulate(it) |
前綴和(或任意二元運算) |
product, permutations, combinations |
組合枚舉 |
count(), cycle(), repeat() |
無限序列,務必配合 islice 或 break |
4. 情境管理器協定
with open(path) as f: 保證離開區塊時呼叫 f.close(),即使區塊裡發生例外。這比 try/finally 少寫一個關閉路徑,也不會漏寫。
自己的資源用類別實作兩個方法:
class Locked:
def __init__(self, lock):
self.lock = lock
def __enter__(self):
self.lock.acquire()
return self # 綁定到 as 後面的名字
def __exit__(self, exc_type, exc, tb):
self.lock.release()
return False # 不吞掉例外
執行順序:
- 計算
with後面的運算式,得到情境管理器cm; - 呼叫
cm.__enter__(),回傳值綁定到as的名字; - 執行區塊;
- 不論區塊是正常結束、
return、break還是例外,都呼叫cm.__exit__(exc_type, exc, tb);沒有例外時三個參數都是None。
__exit__ 回傳真值會吞掉例外,區塊外看不到錯誤。除了明確要忽略的那一種,否則回傳 False(或不回傳)。
檔案、鎖、資料庫交易、暫時改目錄或環境變數、計時,都適合 with。只是為了少寫兩行、卻沒有需要釋放的資源,不必包一層。
5. contextlib 工具箱
contextlib.contextmanager 讓你用產生器寫同一件事:yield 之前是進入,yield 之後是離開。yield 必須放在 try/finally 裡,否則區塊丟出例外時清理不會執行。
from contextlib import contextmanager
import os
@contextmanager
def chdir(path):
old = os.getcwd()
os.chdir(path)
try:
yield path
finally:
os.chdir(old)
with chdir("/tmp"):
print(os.getcwd()) # /tmp
print(os.getcwd()) # 回到原處
| 工具 | 用途 |
|---|---|
contextmanager |
用產生器定義情境管理器 |
closing(obj) |
離開時呼叫 obj.close(),給沒有實作協定的舊物件用 |
suppress(ExcType) |
區塊內指定例外直接忽略 |
redirect_stdout(f) |
暫時把 print 導到檔案 |
ExitStack() |
動態開啟不定數量的資源,離開時逆序全部關閉 |
nullcontext() |
「可選的」情境:條件不成立時給一個什麼都不做的管理器 |
ExitStack 解決「要同時開 N 個檔案,N 在執行期才知道」的問題:
from contextlib import ExitStack
with ExitStack() as stack:
files = [stack.enter_context(open(p)) for p in paths]
# 任何一個 open 失敗,前面已開的都會被關閉
Python 3.10 起 with 可以用括號換行列出多個管理器:
with (
open("a.txt") as a,
open("b.txt") as b,
):
...
6. GIL 實際影響什麼
CPython 預設組建有全域解釋器鎖(GIL):同一個行程、同一時間,只有一個執行緒在執行 Python 位元組碼。執行緒在等待 I/O(socket、磁碟、time.sleep)時會釋放這把鎖,所以多執行緒下載、讀檔仍然有用。
純 Python 的數字迴圈不會釋放鎖,開四個執行緒不會用到四個核心,有時還比單執行緒慢,因為多了切換成本。這時用多行程,或把熱點交給會在計算期間釋放 GIL 的擴充模組(NumPy 的許多運算、hashlib、zlib 都屬於這一類)。
Python 3.13 起提供實驗性的 free-threaded 組建(python3.13t),3.14 繼續改善,但它不是預設值。教材與作業在沒有特別宣告的環境裡,仍應假設 GIL 存在。不要寫出「Python 執行緒不能並行」這種過度的句子:I/O 可以並行,CPU 位元組碼在預設組建裡不行。
量測比記憶規則可靠:
import time, threading
def busy(n):
s = 0
for i in range(n):
s += i
return s
N = 10_000_000
t0 = time.perf_counter(); busy(N); busy(N)
single = time.perf_counter() - t0
t0 = time.perf_counter()
ts = [threading.Thread(target=busy, args=(N,)) for _ in range(2)]
for t in ts: t.start()
for t in ts: t.join()
threaded = time.perf_counter() - t0
print(f"single={single:.2f}s threaded={threaded:.2f}s")
預設組建上兩個數字接近,甚至 threaded 更慢;換成 multiprocessing.Pool(2).map(busy, [N, N]) 才會接近一半。
7. 三種並行怎麼選
| 名稱 | 同時有幾個解釋器在跑 Python | 適合 | 第 3 課對應章節 |
|---|---|---|---|
threading |
預設組建裡,同一時間一個執行緒在執行位元組碼 | 等待 I/O:網路、磁碟、睡眠 | Threading |
asyncio |
仍然一個執行緒,協程在 await 點互相讓出 |
大量連線的 I/O,不想為每條連線開執行緒 | Asyncio |
multiprocessing |
每個行程各自一個解釋器 | CPU 密集的純 Python 計算 | Multiprocessing |
- 工作是等待網路或磁碟,數量不多:
threading(或concurrent.futures.ThreadPoolExecutor),甚至同步寫法就夠。 - 同時成千上萬條連線:
asyncio,並確認所用的函式庫有非同步版本(aiohttp、asyncpg)。 - 要吃滿多核、工作是純 Python 計算:
multiprocessing或ProcessPoolExecutor。把輸入切塊,不要用行程池去跑極小的任務,啟動與序列化成本會蓋過計算。 - 已有 NumPy、資料庫驅動這類會釋放 GIL 或自己用 C 執行緒的函式庫:先量測,再決定要不要多行程。很多時候熱點根本不在 Python 位元組碼。
執行緒共享記憶體,所以要用鎖保護共享的可變物件;行程不共享記憶體,資料要 pickle 傳過去;協程在同一執行緒裡交錯,沒有「兩個 += 同時發生」的資料競爭,但一個協程若寫了很長的 CPU 迴圈、中間沒有 await,其他協程就全部卡住。asyncio.sleep 會把控制權交出去,time.sleep 在協程裡會把整個執行緒睡住。
共享可變狀態能避免就避免。用 queue.Queue / multiprocessing.Queue 傳資料,比用鎖圍住一個大字典容易講對。
8. 常見陷阱
- 產生器當串列重複使用。 第二次是空的。要重複用就
list()一次,或寫成可重新iter()的類別。 groupby沒先排序。 只會合併「相鄰」的相同鍵,結果看起來像 bug。@contextmanager的yield沒包try/finally。 區塊出錯時資源不會釋放。__exit__回傳True。 錯誤消失,測試還是綠的。- 在
async def裡呼叫time.sleep或阻塞式requests。 事件迴圈停擺。用asyncio.sleep與非同步 HTTP 客戶端,或await loop.run_in_executor(...)。 - 用執行緒加速 CPU 迴圈,卻沒量測。 預設 CPython 上通常不會變快。
- 多行程傳遞不能 pickle 的物件。 區域函式、lambda、已開啟的檔案物件、資料庫連線都會在邊界上失敗。
- 鎖的取得順序不一致。 兩個執行緒各拿一把鎖再等對方的另一把,就是死結。所有執行緒用同一順序取得鎖。
9. 練習題
squares = (i*i for i in range(3))。第一次list(squares)與第二次list(squares)各是什麼?- 下面的產生器委派後,
result是多少?def inner(): yield 1 yield 2 return "done" def outer(): result = yield from inner() yield result print(list(outer())) - 為什麼
with open(...) as f在讀到一半丟出例外時,檔案仍會被關閉? - 寫一個
timer()情境管理器,離開區塊時印出經過的秒數;分別用類別與@contextmanager各寫一次。 - 四核機器上,一個純 Python、沒有 I/O 的迴圈,開 4 個執行緒為什麼不能期待 4 倍速度?同一個迴圈若改成每個工作都在等網路回應,執行緒為什麼又有用?
- 協程裡不小心呼叫了
time.sleep(5)。其他已經排進事件迴圈的協程會怎樣?
參考答案
- 第一次是
[0, 1, 4]。第二次是[],產生器已經耗盡。 [1, 2, 'done']。yield from先把inner的 1、2 交出去,inner的return "done"成為yield from的值,再被outer用yield交出。open回傳的檔案物件是情境管理器。離開with時一定呼叫__exit__,其中會關閉檔案,不論離開原因是正常結束、return還是例外。- 類別版:
__enter__記下time.perf_counter()並回傳self;__exit__印出差值、回傳False。產生器版:@contextmanager def timer(): t0 = time.perf_counter() try: yield finally: print(f"{time.perf_counter() - t0:.3f}s") - 預設 CPython 的 GIL 讓這些執行緒輪流執行位元組碼,同一時間大致只有一個核心在跑這段 Python,還要加上切換成本。等待網路時執行緒會釋放 GIL,其他執行緒可以同時在等自己的回應;等待重疊,總時間接近最慢的那一次,而不是全部相加。
time.sleep不會把控制權還給事件迴圈,整個執行緒停 5 秒,其他協程在這 5 秒內都不會被排到。應改成await asyncio.sleep(5)。
小結
迭代器協定只有 __iter__ 與 __next__,產生器用 yield 幫你實作它,並把「下一個值」延到真的被要的時候;itertools 讓這些延遲序列可以切片、串接、分組。情境管理器把取得與釋放寫在同一個 with 裡,__exit__ 不要吞例外,@contextmanager 的 yield 要包 try/finally。執行緒適合重疊 I/O,協程適合大量連線的非阻塞 I/O,行程才適合預設 CPython 上的純 Python 平行計算。選之前先分清工作是在等,還是在算。