迭代器与生成器

对 1000 万行日志做过滤，第一反应经常是：

def filter_timeout_greedy(path: str) -> list[str]:
    with open(path, encoding="utf-8") as f:
        lines = f.readlines()                      # 一千万个 str 进堆
    return [ln for ln in lines if "timeout" in ln]

readlines() 一次分配一个 list，里面一千万个指针，每个指针再指向一个独立的 str。64 位 CPython 上，光 list 槽位大约 8 * 10^7 字节；字符串本体按平均 80 字节算，再加对象头，轻松过 GB。过滤完 lines 还占着，直到函数返回。

生成器管道是另一条路：打开文件、逐行产出、过滤、计数，全程只有「当前这一行」活着。

from collections.abc import Iterable, Iterator


def grep(pattern: str, lines: Iterable[str]) -> Iterator[str]:
    for line in lines:
        if pattern in line:
            yield line


def filter_timeout_lazy(path: str) -> Iterator[str]:
    with open(path, encoding="utf-8") as f:
        yield from grep("timeout", f)              # 文件对象本身就是迭代器

filter_timeout_lazy 返回时文件还开着——这是后文要挖的资源坑。先把内存账算清。下面用 tracemalloc 对比物化和惰性（行数缩到 20 万，数量级关系不变）：

import tracemalloc
from collections.abc import Iterator


def peak_of(factory) -> tuple[int, int]:
    tracemalloc.start()
    n = factory()
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    return peak, n


def greedy(n: int) -> int:
    lines = [f"INFO i={i} error=timeout payload={'x' * 40}\n" for i in range(n)]
    return len([ln for ln in lines if "timeout" in ln])


def lazy(n: int) -> int:
    def lines() -> Iterator[str]:
        for i in range(n):
            yield f"INFO i={i} error=timeout payload={'x' * 40}\n"
    return sum(1 for ln in lines() if "timeout" in ln)


if __name__ == "__main__":
    n = 200_000
    g_peak, g_n = peak_of(lambda: greedy(n))
    l_peak, l_n = peak_of(lambda: lazy(n))
    print(g_n, l_n)
    print("greedy_mb", round(g_peak / 1024 / 1024, 2))
    print("lazy_mb  ", round(l_peak / 1024 / 1024, 2))
    print("ratio    ", round(g_peak / l_peak, 1))

greedy 的峰值跟着 n 线性涨；lazy 的峰值几乎不随 n 变（就一个生成器帧 + 当前字符串）。1000 万行时这个比是几个数量级，不是几个百分点。

C++ 里对应的错误是 std::vector<std::string> lines; while (getline) lines.push_back(...)，正确的是流式读；Go 里是 os.ReadFile / bufio.Scanner 的差别。Python 的陷阱更隐蔽：for line in f 已经是流式的，一行 readlines() 就退回全量物化。生成器不是语法糖，是把「流」变成语言里的一等对象。

---

一、可迭代 vs 迭代器：两套协议

1、名字先分开

日常把「能 for 的东西」统称可迭代。协议上是两层：

角色	需要的方法	行为
Iterable（可迭代对象）	__iter__ 返回一个迭代器	可被多次 for，每次新迭代器
Iterator（迭代器）	__iter__ 返回 self，__next__ 产出下一个	一次性，耗尽后永远停

list / tuple / dict / range / 文件对象，都属于可迭代；list 自己不是迭代器。iter(xs) 才拿到迭代器。

from collections.abc import Iterable, Iterator

xs = [10, 20, 30]
print(isinstance(xs, Iterable), isinstance(xs, Iterator))   # True False

it = iter(xs)           # 等价于 xs.__iter__()
print(isinstance(it, Iterable), isinstance(it, Iterator))   # True True
print(it is xs)         # False，list_iterator 是另一个对象
print(next(it), next(it), next(it))
try:
    next(it)
except StopIteration as e:
    print("exhausted", e.value)     # value 默认 None

collections.abc.Iterator 是 Iterable 的子类：迭代器一定可迭代（__iter__ 返回自己），反过来不成立。

C++ 的 iterator 是一份「指向容器某处」的值，多半可拷贝；vector::iterator 属于 RandomAccess，同一份容器上可以同时活很多个。Go 1.23 之前 range 是语言内置的，slice/map/channel 各走各的；1.23 的 iter.Seq[T] 才接近 Python 这套协议。Python 把协议暴露成方法，所以你可以自己实现，也可以用错。

2、iter() 的两种重载

单参 iter(obj)：先找 obj.__iter__；没有则退化到「从 0 开始 __getitem__，直到 IndexError」。这是历史兼容，自己写类型时不要靠这条暗路，显式给 __iter__。

双参 iter(callable, sentinel)：反复调用 callable()，直到返回值 == sentinel。这是把「带终止条件的读」收成迭代器的标准手法：

from io import BytesIO

buf = BytesIO(b"abcdefghij")
chunks = iter(lambda: buf.read(3), b"")
print(list(chunks))          # [b'abc', b'def', b'ghi', b'j']

文件分块、socket 读到空、队列里取到哨兵，都是这个模式。C++ 没有直接对应物，手写 while (true) { auto n = read(); if (!n) break; }；Go 是 io.ReadFull / bufio 循环。Python 把它收进 for chunk in iter(...)。

3、for 的反糖

for x in xs:
    body

展开以后是：

_it = iter(xs)
while True:
    try:
        x = next(_it)
    except StopIteration:
        break
    body

没有下标，没有 len，不要求随机访问。这就是为什么 for line in f 能处理比内存大的文件，也是为什么无限流（itertools.count）能写进 for——只要你自己 break。

StopIteration 是协议内部信号，不要在业务代码里靠抛 StopIteration 来跳出多层循环。从 Python 3.7 起（PEP 479），生成器函数内部冒出来的 StopIteration 会被包成 RuntimeError，防止它悄悄终止外层生成器。要停，用 return；要向外传「没有了」，让 __next__ 自然耗尽。

def bad() -> Iterator[int]:
    raise StopIteration(42)     # 3.7+ 变成 RuntimeError


def good() -> Iterator[int]:
    yield 1
    return 42                   # 值挂在 StopIteration.value 上，yield from 能接到

4、迭代器是一次性的，list 可反复

xs = [1, 2, 3]
print(list(xs), list(xs))       # 两次都是 [1, 2, 3]

it = iter(xs)
print(list(it), list(it))       # [1, 2, 3] 然后 []

list.__iter__ 每次造新的 list_iterator，内部一个游标，不改 list 本身。所以 for a in xs: for b in xs: 是笛卡尔积。文件对象相反：__iter__ 返回 self，读指针在对象上，第二次 for 从上次停下的地方继续，通常是空。

from io import StringIO

f = StringIO("a\nb\nc\n")
print(list(f), list(f))         # ['a\n', 'b\n', 'c\n'] 然后 []

设计自己的类型时，这是第一件要决定的事：

• 可反复遍历（容器）：__iter__ 每次返回新迭代器，状态放在迭代器上，不放在容器上。
• 一次性流（文件、套接字、生成器）：对象自己就是迭代器，__iter__ 返回 self。

把流写成容器，第二次循环是空的，bug 从「偶发」变成「必现」。把容器写成流，嵌套 for 直接坏。

C++ 里 istream_iterator 是 InputIterator，单遍；vector::iterator 可多遍。Python 没有 iterator category，只有「你自己记得它是哪一种」。Go 的 channel range 是单遍且破坏性的（值被取走）；slice range 是多遍。

5、为什么 dict 迭代时不能改 size

d = {"a": 1, "b": 2, "c": 3}
try:
    for k in d:
        d[k + k] = 0            # 插入，size 变了
except RuntimeError as e:
    print(e)                    # dictionary changed size during iteration

CPython 的 dict 迭代器在内部记下了开始时的 ma_used（条目数）。每次 next 先比对，变了就抛。原因不是「hash 表不能遍历」，而是：插入可能触发扩容、重排 compact 表，迭代器手里的偏移量立刻失效；即使没扩容，新 key 插在当前游标前还是后，是「跳过」还是「重复」都没有合理语义。解释器选择直接拒绝。

改已有 key 的值不改 size，允许：

d = {"a": 1, "b": 2, "c": 3}
for k in d:
    d[k] += 1
print(d)                        # {'a': 2, 'b': 3, 'c': 4}

删当前 key 同样改 size，同样抛。要边遍历边删，先快照：

d = {"a": 1, "b": 2, "c": 3, "skip": 0}
for k in list(d):               # 独立的 key 列表
    if k == "skip":
        del d[k]
print(d)

list(d) 是 O(n) 内存，换来语义确定。dict.keys() 是动态视图，不能当快照用。

set 同一套规则。list 更阴：迭代时 append 不会抛，游标按索引走，你在尾巴上加东西就是无限循环。

xs = [1, 2, 3]
n = 0
for x in xs:
    n += 1
    if n < 6:
        xs.append(x)            # 别在生产里这么写
    else:
        break
print(xs, n)

Go 对 map 的 range：迭代中删除当前 key 是允许的，新增 key 是否被看到未定义。C++ 对 unordered_map 插入可能使所有 iterator 失效。Python 用 RuntimeError 把「未定义」变成「必崩」，调试上更友好，前提是你知道它会崩。

---

二、生成器函数：yield 把栈帧冻住

1、yield 不是 return

函数体里出现 yield，调用它不会执行函数体，只制造一个生成器对象。函数体在第一次 next 时才跑到第一个 yield，把值交出，栈帧挂起——局部变量、指令指针、求值栈全部冻在生成器对象上。

def countdown(n: int) -> Iterator[int]:
    print("start", n)
    while n > 0:
        yield n
        n -= 1
    print("end")
    return "done"


g = countdown(3)
print(type(g))                  # <class 'generator'>
print(g is iter(g))             # True，生成器是迭代器
print("not started yet")
print(next(g), next(g), next(g))
try:
    next(g)
except StopIteration as e:
    print("return value:", e.value)

输出顺序是：not started yet → start 3 → 3 2 1 → end → return value: done。return 的值不通过 for 拿到（for 吞掉 StopIteration），只挂在异常的 .value 上。yield from 能接到这个值，普通 for 不能。

C++23 的 std::generator、C# 的 yield return、JS 的 function* 是同一类东西。Go 没有生成器；习惯用 channel + goroutine 模拟惰性流，但那是并发原语，有缓冲、有生命周期、有泄漏模型，和冻栈帧不是一回事。

2、状态机就藏在对象上

import inspect


def gen() -> Iterator[int]:
    yield 1
    yield 2


g = gen()
print(inspect.getgeneratorstate(g))     # GEN_CREATED
next(g)
print(inspect.getgeneratorstate(g))     # GEN_SUSPENDED
list(g)
print(inspect.getgeneratorstate(g))     # GEN_CLOSED

四个状态：GEN_CREATED / GEN_RUNNING / GEN_SUSPENDED / GEN_CLOSED。GEN_RUNNING 只能在生成器自己的栈上看到（或调试器里）；生成器不能重入——gi_running 时再 next/send 会 ValueError: generator already executing。

挂起时局部变量还活着，所以生成器是闭包之外另一种「带状态的函数」。也因此，它持有的引用（大 list、锁、文件、socket）在耗尽或 close 之前不会释放——不是 GC 的 bug，是帧还在。g.gi_frame 非空就能在 f_locals 里看到那些名字；close 之后帧拆掉。

3、yield 表达式可以 send

PEP 342 让 yield 变成表达式：外面 g.send(x) 的值，成为生成器内部 yield 的结果。第一次必须 next(g) 或 g.send(None)，因为还没有任何 yield 在等值。

from collections.abc import Generator


def running_avg() -> Generator[float, float, None]:
    """yield 产出当前均值，send 喂新样本。"""
    total = 0.0
    n = 0
    avg = 0.0
    while True:
        x = yield avg
        if x is None:
            continue
        total += x
        n += 1
        avg = total / n


g = running_avg()
print(next(g))              # 0.0，跑到 yield avg
print(g.send(10.0))         # 10.0
print(g.send(20.0))         # 15.0
print(g.send(30.0))         # 20.0
g.close()

类型是 Generator[YieldType, SendType, ReturnType]。只产出、不接收，写成 Iterator[YieldType] 更准确。send 让生成器看起来像协程：调用方和生成器互相交接。这就是「yield 协程」这条旧路的全部家当——单线程协作，没有 IO 多路复用，没有事件循环。现在不要用它写并发，第六节再对照。

4、throw / close

g.throw(exc) 在当前 yield 点注入异常。生成器可以捕获、处理、继续 yield，也可以让异常冒泡，冒泡时生成器关闭。

def guarded() -> Iterator[str]:
    try:
        yield "ready"
        yield "still"
    except ValueError as e:
        yield f"caught {e}"
    yield "after"


g = guarded()
print(next(g))
print(g.throw(ValueError("boom")))
print(next(g))

g.close() 注入 GeneratorExit（继承 BaseException，不是 Exception）。清理完让它继续传播，或直接 return。禁止捕获后再 yield：会报 RuntimeError: generator ignored GeneratorExit。GC 时 gen_del 也会 close，所以「拿了锁没耗尽就丢掉」不一定死锁，但时机不确定，不能当释放协议。要确定释放，用 try/finally 或 with。

5、生成器表达式 vs 列表推导

基础篇说过内存差异。这里补协议层面的后果。

xs = [1, 2, 3, 4, 5]
squares_list = [x * x for x in xs]
squares_gen = (x * x for x in xs)

print(type(squares_list), type(squares_gen))
print(squares_gen.gi_frame.f_locals)    # 生成器表达式也有帧
print(sum(squares_gen), sum(squares_gen))  # 25 然后 0，只能走一遍

生成器表达式的作用域是独立的（3.x 起推导式也不再泄漏循环变量）。它捕获迭代源的引用：

def make() -> tuple[list[int], Iterator[int]]:
    data = [1, 2, 3]
    return data, (x for x in data)


data, g = make()
data.append(4)
print(list(g))                  # [1, 2, 3, 4]，源是同一个 list

这和 C++ 里「对 vector 取了 iterator 之后 push_back 可能失效」同类。生成器表达式没有拷贝源。源如果是一次性迭代器，表达式会把它耗尽：(x * 2 for x in iter([1, 2, 3])) 之后原 iter 已经空了。

什么时候用哪一个：

• 只扫一遍、下游是 sum / any / all / max / 另一个生成器：生成器表达式。
• 要下标、要 len、要传给多次遍历的 API、要切片：list。
• 不要写 list((x for x in xs))，那就是 [x for x in xs]，还多一层。
• 不要对生成器做 xs[10:20]，先 list 再切等于前 20 个全物化。用 itertools.islice。

---

三、yield from：委托，不是语法糖 for

1、为什么不能只写 for x in sub: yield x

表面像：

def flatten_naive(nested: Iterable[Iterable[int]]) -> Iterator[int]:
    for sub in nested:
        for x in sub:
            yield x

yield from sub 看起来只是少一层循环。真正多出来的是 双向委托：send、throw、close、子生成器的 return 值，全部由 yield from 转发。手写 for 只转发了产出，把生成器管道的控制信道掐断了。

from collections.abc import Generator


def sub() -> Generator[int, int, str]:
    acc = 0
    while True:
        x = yield acc
        if x is None:
            break
        acc += x
    return "sub-done"


def wrapper() -> Generator[int, int, str]:
    result = yield from sub()
    return f"wrapper got {result}"


g = wrapper()
print(next(g))                  # 0
print(g.send(5))                # 5
print(g.send(7))                # 12
try:
    g.send(None)
except StopIteration as e:
    print(e.value)              # wrapper got sub-done

yield from 把 wrapper 的 send(5) 原样交给 sub 里那个 yield acc。子生成器 return "sub-done" 变成 yield from 表达式的值。PEP 380 的展开有几十行，核心就是：当前 yield from 挂起时，生成器的 gi_yieldfrom 指向子迭代器，所有通道往下灌。

对普通可迭代对象（list、文件），yield from xs 等价于 for x in xs: yield x，因为它们不接收 send。对生成器，两者不等价。写管道时一律 yield from，少想。

2、生成器管道

日志那条链完整写出来：

from collections.abc import Iterable, Iterator
from pathlib import Path


def open_lines(path: Path) -> Iterator[str]:
    with path.open(encoding="utf-8", errors="replace") as f:
        yield from f                           # 文件关闭绑定在这个生成器的 finally


def strip_newlines(lines: Iterable[str]) -> Iterator[str]:
    for ln in lines:
        yield ln.rstrip("\n")


def not_comment(lines: Iterable[str]) -> Iterator[str]:
    for ln in lines:
        if ln and not ln.startswith("#"):
            yield ln


def parse_kv(lines: Iterable[str]) -> Iterator[tuple[str, str]]:
    for ln in lines:
        if "=" not in ln:
            continue
        k, _, v = ln.partition("=")
        yield k.strip(), v.strip()


def pipeline(path: Path) -> Iterator[tuple[str, str]]:
    return parse_kv(not_comment(strip_newlines(open_lines(path))))

每一节输入可迭代、输出迭代器，没有中间 list。pipeline 本身还没读文件——直到下游 for / list / any 拉动。Unix 管道是进程 + 字节流；这里是栈帧 + Python 对象。拉不动就不算，这是惰性的定义。

中间如果写了 list(...)，管道在那一层断开，内存重新按全量涨。审查生成器代码，搜 list( 和 [] 推导式。空的 yield from () 直接往下走；yield from None 是 TypeError，可选下游用空 tuple，不要传 None。

---

四、itertools：惰性工具箱和它的代价

itertools 里的函数几乎都返回迭代器。它们不比手写 for 快几个数量级（还是 Python 层调用），价值是语义短、组合时不物化。C++ 对应 <ranges> / <iterator>，Go 对应 slices 加上 1.23 的 iter。

import itertools as it

1、islice：对迭代器切片

g = (i for i in range(1000))
print(list(it.islice(g, 10, 20)))       # [10, 11, ..., 19]
print(next(g))                          # 20，前面的被消耗掉了

islice(iterable, start, stop, step) 不支持负数下标——迭代器不能从尾巴数。list(g)[10:20] 会把 g 全部拉进内存再切；islice 只丢弃前 10 个、保留 10 个。无限流上这是唯一合法的「切片」。

step 靠跳过实现，不是随机访问。对 list 做 islice 没有比切片更快，别装。

2、chain / chain.from_iterable

print(list(it.chain([1, 2], (3, 4), "ab")))          # [1, 2, 3, 4, 'a', 'b']
print(list(it.chain.from_iterable([[1, 2], [3], [4, 5]])))

chain(*iterables) 把参数元组钉在栈上，可迭代对象很多时用 chain.from_iterable，一次传一个可迭代的可迭代。拼接大量 list 不要 sum(lists, [])（平方），不要反复 +，chain.from_iterable 再视需要 list(...)。

3、groupby：连续分组，不是 SQL GROUP BY

rows = [
    ("us-east", "a"),
    ("us-east", "b"),
    ("eu-west", "c"),
    ("us-east", "d"),           # 又回到 us-east
]
for key, group in it.groupby(rows, key=lambda r: r[0]):
    print(key, list(group))

输出三段：us-east 两条、eu-west 一条、us-east 一条。groupby 只把相邻的相同 key 收成一组。要全局聚合，先 sorted(rows, key=...) 再 groupby，或者老实用 dict[list]。

更阴的一条：group 本身是迭代器，共享底层源。进入下一组时上一组失效。

rows = [("a", 1), ("a", 2), ("b", 3)]
saved = []
for key, group in it.groupby(rows, key=lambda r: r[0]):
    saved.append((key, group))          # 没立刻 list()
print([(k, list(g)) for k, g in saved]) # 第一组空了

要用就当场 list(group)。这不是 bug，是「零拷贝连续扫描」的代价。

4、tee 会缓存，不是免费分叉

src = (i for i in range(5))
a, b = it.tee(src, 2)
print(list(a))
print(list(b))                  # 还能拿到 [0,1,2,3,4]

看起来 tee 把一次性迭代器变成了两个。实现是：共享一个 deque，领先的那边把值推进缓冲，落后的从缓冲取。若 a 跑完、b 才开始，deque 里就是全部历史，内存退化成 list(src)，还多一层间接。

def peek_cost() -> None:
    src = (i for i in range(100_000))
    a, b = it.tee(src, 2)
    _ = list(a)                 # 领先到底
    # 此时 tee 的缓冲 ≈ 十万个引用
    _ = list(b)


peek_cost()

tee 的正当用途：需要「看一眼下一个再决定」且窗口很小。两个下游以相近速率消费，缓冲保持 O(窗口)。已经有 list 就别 tee，直接用两次。tee 之后不要再碰原来的 src。

count / cycle / repeat 能造无限流，必须用 islice 或自己 break 截断。cycle 内部会把可迭代物化成 list 再循环——传入生成器等于偷偷 list()，再一次缓存代价。batched（3.12+）按长度切块，accumulate 做前缀和，都是同一类惰性组合件。

---

五、自定义可迭代类

生成器函数够用时不要上类。类的价值是：可反复遍历、带配置、可以同时是上下文管理器、需要 __len__ / __getitem__。

1、文件分块读取

一次性流，对象自己当迭代器也可以；要「同一个路径 for 两次都从头读」，__iter__ 每次新开文件。

from collections.abc import Iterator
from pathlib import Path


class ChunkedFile:
    def __init__(self, path: Path, size: int = 8192) -> None:
        if size <= 0:
            raise ValueError("size must be positive")
        self.path = path
        self.size = size

    def __repr__(self) -> str:
        return f"ChunkedFile({self.path!r}, size={self.size})"

    def __iter__(self) -> Iterator[bytes]:
        # yield 让 __iter__ 变成生成器函数：每次 for 一个新生成器、一扇新文件
        with self.path.open("rb") as f:
            yield from iter(lambda: f.read(self.size), b"")

list(ChunkedFile(path, 4)) 两次都会从头读：每次 for 一个新生成器、一扇新文件。

对比错误写法：在 __init__ 里 open，__iter__ 返回 self，__next__ 读一块。第一次 for 结束文件在 EOF，第二次空；异常路径还要把文件关在 __del__ 里。能跑，但生命周期绑错了对象。

2、滑动窗口

from collections import deque
from collections.abc import Iterable, Iterator
from typing import TypeVar

T = TypeVar("T")


class SlidingWindow(Iterable[tuple[T, ...]]):
    def __init__(self, source: Iterable[T], n: int) -> None:
        if n <= 0:
            raise ValueError("n must be positive")
        self._source = source
        self._n = n

    def __iter__(self) -> Iterator[tuple[T, ...]]:
        buf: deque[T] = deque(maxlen=self._n)
        for x in self._source:
            buf.append(x)
            if len(buf) == self._n:
                yield tuple(buf)


print(list(SlidingWindow(range(5), 3)))
# [(0, 1, 2), (1, 2, 3), (2, 3, 4)]

deque(maxlen=n) 丢左边是 O(1)。每次 yield tuple(buf) 是一次 O(n) 拷贝，换来窗口不被后续 append 改掉——调用方拿着元组是安全的。若产出 buf 本身，下游稍一持有，下一轮窗口已经变了。

源是生成器时，SlidingWindow 只能走一遍，因为状态在源上不在窗口对象上。要反复滑，源得是容器。把这个写进 docstring，比加一个假的「可反复」更有用。

3、__iter__ 返回 self 的时候必须实现 __next__

class Counter:
    def __init__(self, n: int) -> None:
        self.n = n
        self.i = 0

    def __iter__(self) -> "Counter":
        return self

    def __next__(self) -> int:
        if self.i >= self.n:
            raise StopIteration
        self.i += 1
        return self.i - 1


c = Counter(3)
print(list(c), list(c))         # [0, 1, 2] 然后 []

这种对象是迭代器。要让它可反复，游标不能放在 self.i 上：

class CounterIterable:
    def __init__(self, n: int) -> None:
        self.n = n

    def __iter__(self) -> Iterator[int]:
        return Counter(self.n)  # 每次新迭代器

或者直接在 __iter__ 里 yield，少一个类。面试里手写迭代器协议有用；生产里生成器函数是默认选项。

只实现 __getitem__ 也能被 for 驱动（从 0 收到 IndexError），那是协议的退化路径，iter() 每次造一个 iterator 包装。缺 __len__ 时 list(obj) 不知道预分配多大，靠 __length_hint__ 可选地提示。自己写容器，__iter__ + __len__ 给齐。

---

六、生成器 vs 协程：这里停在分界线上

send 让生成器能接收值，PEP 342 把它叫协程。asyncio 早期就是这么干的：result = yield from some_future。问题有一串：调用方必须知道谁在 send；忘记驱动就饿死；yield 既表示「产出一个业务值」又表示「等 IO」；堆栈在生成器链上，错误栈难看。

PEP 492 把这件事拆开：

• 生成器：yield / yield from，同步拉数据。
• 协程：async def / await，由事件循环驱动。
• 异步生成器：async def + yield，async for 消费（细节留给并发篇）。

分界规则：没有 IO 等待、只是不想物化中间结果 → 生成器。有 socket / sleep / 文件异步 IO → async def。不要用 yield 协程去模拟 await，也不要把 async def 当生成器用（调用得到 coroutine 对象，不迭代它会告警 was never awaited）。

C++ 的 coroutine（co_await / co_yield）和 Python 的 async 更近，和同步 yield 生成器不是一个讨论。Go 的 goroutine 是抢占/协作混合的并发单元，更不是生成器。把三个词混在简历上，面试第一问就会拆。

---

七、会踩的坑：生成器里持有锁和文件

1、yield 在 try 里把 finally 推迟到未知时刻

def leaky_lines(path: str) -> Iterator[str]:
    f = open(path, encoding="utf-8")
    for line in f:
        yield line
    f.close()                   # 只有正常耗尽才到达


def leaky_locked(items: list[int], lock: "threading.Lock") -> Iterator[int]:
    lock.acquire()
    for x in items:
        yield x
    lock.release()              # break / return / 异常都跳过

调用方写 for line in leaky_lines(p): if cond: break，文件句柄和锁都留在生成器帧上。帧被 GC 时 close() 会注入 GeneratorExit，for 循环被打断，f.close() 仍可能没执行——leaky_lines 把 close 放在循环后面，不是 finally。锁更糟：另一根线程永远拿不到。

import gc
import threading

lock = threading.Lock()


def hold(items: list[int]) -> Iterator[int]:
    lock.acquire()
    for x in items:
        yield x
    lock.release()


g = hold([1, 2, 3])
print(next(g), lock.locked())   # 1 True
del g
gc.collect()
print("after gc", lock.locked())  # 可能已释放，时机不确定

「可能已释放」不是「已经释放」。最终化线程、循环引用、解释器关机顺序，都会让 __del__ / gen.close 迟到或根本不来。资源协议必须是确定性的。

2、try/finally 绑在生成器帧上，仍然要人把它关掉

def lines(path: str) -> Iterator[str]:
    f = open(path, encoding="utf-8")
    try:
        for line in f:
            yield line
    finally:
        f.close()

finally 在三种情况下跑：正常耗尽、g.close()、GeneratorExit（含 GC）。break 出 for 时，for 会对迭代器调 close（如果有）——生成器有 close，所以 for line in lines(p): break 会进 finally。这是 3.x 的保证。

但把生成器交给一个只调 next 几次的人，或者塞进 zip / islice 提前停、而外层没有 close，就不一定。自己暴露生成器 API 时，要么文档写清「必须耗尽或 close」，要么不要在生成器里持有独占资源。

3、with 包住 yield from，让生命周期跟生成器走

from pathlib import Path
import threading


def lines(path: Path) -> Iterator[str]:
    with path.open(encoding="utf-8") as f:
        yield from f            # 退出这个生成器（耗尽/close/异常）才离开 with


def iter_under_lock(lock: threading.Lock, items: list[int]) -> Iterator[int]:
    with lock:
        yield from items        # 消费完成（或 close）才解锁；临界区会被拉得很长


def iter_snapshot(lock: threading.Lock, items: list[int]) -> Iterator[int]:
    with lock:
        snap = tuple(items)
    yield from snap             # 锁外产出，持有的是快照

yield inner() 把迭代器交出去、with 已经结束，锁释放了调用方还在迭代——生命周期错位。独占资源要和消费同范围。调用方必须在同一个线程里把迭代器拉完。锁跨 yield 通常不是你想要的；多数时候该快照再在锁外迭代。

文件适合「生成器持有句柄」；锁通常不适合。默认选项是：文件用 with + yield from；锁用短临界区 + 快照。生成器自己有 close，提前停用 contextlib.closing(g) 或显式 g.close()，帧才会拆掉。自己写的流类型优先实现 __enter__ / __exit__（下一篇）。

---

八、工程上怎么选

协议记住三句话：

1. 可迭代对象制造迭代器；迭代器是一次性游标；for 只认识这套。
2. 生成器是编译器写好的迭代器，帧冻在 yield；send/throw/close/yield from 是控制信道。
3. 惰性省内存的前提是真的只走一遍、中间不物化、资源跟着帧走。

对照着选：

需求	选
多次遍历、要 len / 下标	list / 自定义容器
单遍过滤、流水线、大文件	生成器函数 / 生成器表达式
对无限流或大流切片	islice，禁止先 list
分叉同一条流	先问窗口多大；大就 list，小才 tee
分组	相邻用 groupby，全局用 dict
文件句柄	with open 包住 yield from
锁	短临界区快照，不要跨 yield 持锁
等 IO	停，去 async for，不要 send

C++ 程序员容易把生成器当惰性 vector，然后 &g[0] 式地用第二次。Go 程序员容易把生成器当无缓冲 channel，然后在另一个 goroutine 里拉——生成器不能跨线程安全地 next。它就是当前线程上一个冻住的栈帧。把这个模型用到底，管道、泄漏、tee 缓存、dict 的 RuntimeError 都是同一张图上的边。