進階前九課已經涵蓋物件導向、函數式寫法、並行、資料處理、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. 迭代器協定
  2. 產生器函式與 yield
  3. 產生器管線與 itertools
  4. 情境管理器協定
  5. contextlib 工具箱
  6. GIL 實際影響什麼
  7. 三種並行怎麼選
  8. 常見陷阱
  9. 練習題

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)) 的差別。

三個常被忽略的細節:

  1. 產生器只能走一次。 走完再迭代,不會重新執行函式,而是什麼都沒有。需要第二次,就再呼叫一次函式。
  2. return value 會變成 StopIteration.value。 用 yield from 委派時,子產生器的回傳值就是 yield from 運算式的值。
  3. 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           # 不吞掉例外

執行順序:

  1. 計算 with 後面的運算式,得到情境管理器 cm;
  2. 呼叫 cm.__enter__(),回傳值綁定到 as 的名字;
  3. 執行區塊;
  4. 不論區塊是正常結束、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
  1. 工作是等待網路或磁碟,數量不多:threading(或 concurrent.futures.ThreadPoolExecutor),甚至同步寫法就夠。
  2. 同時成千上萬條連線:asyncio,並確認所用的函式庫有非同步版本(aiohttp、asyncpg)。
  3. 要吃滿多核、工作是純 Python 計算:multiprocessing 或 ProcessPoolExecutor。把輸入切塊,不要用行程池去跑極小的任務,啟動與序列化成本會蓋過計算。
  4. 已有 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. 練習題

  1. squares = (i*i for i in range(3))。第一次 list(squares) 與第二次 list(squares) 各是什麼?
  2. 下面的產生器委派後,result 是多少?
    def inner():
        yield 1
        yield 2
        return "done"
    
    def outer():
        result = yield from inner()
        yield result
    
    print(list(outer()))
  3. 為什麼 with open(...) as f 在讀到一半丟出例外時,檔案仍會被關閉?
  4. 寫一個 timer() 情境管理器,離開區塊時印出經過的秒數;分別用類別與 @contextmanager 各寫一次。
  5. 四核機器上,一個純 Python、沒有 I/O 的迴圈,開 4 個執行緒為什麼不能期待 4 倍速度?同一個迴圈若改成每個工作都在等網路回應,執行緒為什麼又有用?
  6. 協程裡不小心呼叫了 time.sleep(5)。其他已經排進事件迴圈的協程會怎樣?

參考答案

  1. 第一次是 [0, 1, 4]。第二次是 [],產生器已經耗盡。
  2. [1, 2, 'done']。yield from 先把 inner 的 1、2 交出去,inner 的 return "done" 成為 yield from 的值,再被 outer 用 yield 交出。
  3. open 回傳的檔案物件是情境管理器。離開 with 時一定呼叫 __exit__,其中會關閉檔案,不論離開原因是正常結束、return 還是例外。
  4. 類別版:__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")
  5. 預設 CPython 的 GIL 讓這些執行緒輪流執行位元組碼,同一時間大致只有一個核心在跑這段 Python,還要加上切換成本。等待網路時執行緒會釋放 GIL,其他執行緒可以同時在等自己的回應;等待重疊,總時間接近最慢的那一次,而不是全部相加。
  6. time.sleep 不會把控制權還給事件迴圈,整個執行緒停 5 秒,其他協程在這 5 秒內都不會被排到。應改成 await asyncio.sleep(5)。

小結

迭代器協定只有 __iter__ 與 __next__,產生器用 yield 幫你實作它,並把「下一個值」延到真的被要的時候;itertools 讓這些延遲序列可以切片、串接、分組。情境管理器把取得與釋放寫在同一個 with 裡,__exit__ 不要吞例外,@contextmanager 的 yield 要包 try/finally。執行緒適合重疊 I/O,協程適合大量連線的非阻塞 I/O,行程才適合預設 CPython 上的純 Python 平行計算。選之前先分清工作是在等,還是在算。