Python►中级进阶

生成器 yield 深入

上一篇 迭代器与可迭代对象 讲了迭代器协议，也提到”写迭代器最优雅的方式是生成器”。这一篇要把 yield、生成器函数、生成器表达式、yield from、协程用法一次讲透。学完你会发现：许多”看起来要一大堆循环、列表”的问题，用生成器一句话就能解决。

一、什么是生成器

只要函数里有 yield，它就是生成器函数——调用它不会立刻执行函数体，而是返回一个生成器对象（一种迭代器）。

1
2
3
4
5
6
7
8
9
10
11
12
def count_down(n):
    while n > 0:
        yield n
        n -= 1

g = count_down(3)
print(g)                  # <generator object count_down at ...>

print(next(g))            # 3
print(next(g))            # 2
print(next(g))            # 1
print(next(g))            # StopIteration

或者直接 for 循环：

1
2
for x in count_down(3):
    print(x)      # 3 2 1

二、yield 与 return

  • return 结束函数，返回一个值；
  • yield 暂停函数，产出一个值，下次调用 next 时从暂停处继续。

关键：生成器函数的局部状态在 yield 之间保留——变量、循环、if 判断都保持原样。

1
2
3
4
5
6
7
8
9
10
11
12
def demo():
    print("A")
    yield 1
    print("B")
    yield 2
    print("C")


g = demo()
print(next(g))    # A → yield 1
print(next(g))    # B → yield 2
print(next(g))    # C → StopIteration

三、生成器表达式

前面 列表推导式与生成表达式 讲过：

1
2
3
4
5
6
7
8
# 列表推导式：立即算出全部
squares_list = [x*x for x in range(10)]

# 生成表达式：惰性
squares_gen = (x*x for x in range(10))

print(next(squares_gen))    # 0
print(next(squares_gen))    # 1

生成表达式内存友好，处理大数据首选。

作为函数唯一参数时括号可省：

1
2
sum(x*x for x in range(1_000_000))   # 内存 O(1)
max(u.age for u in users)

四、生成器的核心价值：惰性

一次只产生一个值 → 内存友好 + 早退可能。

1
2
3
4
5
6
7
8
9
10
11
def read_large_file(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line.rstrip()


# 处理一个 100GB 的日志文件
for line in read_large_file("huge.log"):
    if "ERROR" in line:
        print(line)
        # 想早退随时 break

不用一次读进内存，也不用先造一个大列表。

五、yield from：委托给另一个生成器

1
2
3
4
5
6
7
8
9
10
11
def sub():
    yield 1
    yield 2

def main():
    yield 0
    yield from sub()      # 委托：把 sub 产出的一次性 yield 出去
    yield 3


list(main())     # [0, 1, 2, 3]

yield from x 等价于：

1
2
for v in x:
    yield v

但功能远超简单循环——它还转发 send、throw、返回值。用来”拆分复杂生成器”、”扁平化嵌套”非常好用。

例子：扁平化任意深度嵌套

1
2
3
4
5
6
7
8
9
def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)     # 递归 yield
        else:
            yield item


list(flatten([1, [2, [3, [4, 5]]], 6]))   # [1, 2, 3, 4, 5, 6]

六、生成器 = 惰性数据管道

生成器可以像 Unix 管道一样串联：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
def read_lines(path):
    with open(path, encoding="utf-8") as f:
        for line in f:
            yield line

def filter_errors(lines):
    for line in lines:
        if "ERROR" in line:
            yield line

def parse(lines):
    for line in lines:
        parts = line.split()
        yield {"time": parts[0], "msg": " ".join(parts[1:])}


# 链起来
pipeline = parse(filter_errors(read_lines("huge.log")))
for event in pipeline:
    print(event)

每一步都不占额外内存——线性时间、常数空间。这是生成器的杀手锏。

七、生成器可以 send 数据回去（协程）

yield 不只是”产出”，还能”接收”外部传入的数据：

1
2
3
4
5
6
7
8
9
10
def echo():
    while True:
        received = yield
        print("收到：", received)


g = echo()
next(g)               # 启动：跑到第一个 yield 处停下
g.send("hi")          # 收到：hi
g.send("hello")       # 收到：hello

规则：

  • next(g) 或 g.send(None) 让生成器跑到下一个 yield；
  • g.send(v) 把 v 塞给最近一次 yield 的返回值；
  • 第一次一定要用 next(g) 启动（或 g.send(None)）。

这就是协程的基础。不过现代 Python 里协程一般用 async/await（见 异步编程 asyncio 入门），yield 协程用得少了。

八、生成器可以有 return

return 让生成器结束，返回值放在 StopIteration 里：

1
2
3
4
5
6
7
8
9
10
11
12
13
def gen():
    yield 1
    yield 2
    return "done"


g = gen()
next(g)          # 1
next(g)          # 2
try:
    next(g)      # StopIteration
except StopIteration as e:
    print(e.value)     # 'done'

配合 yield from 可以把这个值传出去：

1
2
3
4
5
6
def outer():
    result = yield from gen()
    print(f"内层返回：{result}")


list(outer())    # [1, 2] + 打印"内层返回：done"

九、close 与 throw

1
2
3
4
5
g = count_down(10)
next(g)
g.close()        # 关闭生成器，触发 GeneratorExit

g.throw(ValueError, "错")     # 在生成器内部抛异常

日常业务代码用得少，主要在写协程/异步框架时才用。

十、生成器 vs 迭代器

  • 写法：生成器（yield）比手写 __iter__ / __next__ 简洁一个数量级；
  • 性能：几乎等同；
  • 能力：完全一样。

能用生成器就别写迭代器类。

十一、性能与陷阱

一次性

跟迭代器一样，生成器只能遍历一次：

1
2
3
g = (x*x for x in range(5))
list(g)     # [0, 1, 4, 9, 16]
list(g)     # []

不能索引

1
2
g = (x for x in range(5))
g[0]     # ❌ TypeError

要索引先 list(g)。

不能 len

1
len((x for x in range(5)))    # ❌

要长度先 list(g) 或自己累加计数。

十二、几个必会的生成器模式

无限序列

1
2
3
4
5
6
7
8
9
def naturals():
    n = 1
    while True:
        yield n
        n += 1


from itertools import islice
list(islice(naturals(), 10))    # [1..10]

分块

1
2
3
4
5
6
7
8
9
from itertools import islice

def chunks(seq, n):
    it = iter(seq)
    while chunk := list(islice(it, n)):
        yield chunk


list(chunks(range(10), 3))    # [[0,1,2], [3,4,5], [6,7,8], [9]]

滑动窗口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from collections import deque

def window(seq, size):
    it = iter(seq)
    buf = deque(islice(it, size), maxlen=size)
    if len(buf) < size:
        return
    yield tuple(buf)
    for x in it:
        buf.append(x)
        yield tuple(buf)


list(window([1, 2, 3, 4, 5], 3))
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]

去重（保序）

1
2
3
4
5
6
7
8
9
def dedupe(seq):
    seen = set()
    for x in seq:
        if x not in seen:
            seen.add(x)
            yield x


list(dedupe([1, 2, 2, 3, 1, 4]))    # [1, 2, 3, 4]

十三、一个完整例子：文件按事件切分

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
# events.py
"""从日志流里按'空行分隔'的规则切分事件。"""


def read_events(lines):
    """lines 是行迭代器；产出一批批事件（一批 = 空行前的所有行）。"""
    buffer = []
    for line in lines:
        line = line.rstrip()
        if not line:
            if buffer:
                yield buffer
                buffer = []
        else:
            buffer.append(line)
    if buffer:
        yield buffer


def enrich(events):
    """给每个事件加上一个 id。"""
    for i, ev in enumerate(events, 1):
        yield {"id": i, "lines": ev}


def keep_errors(events):
    """只保留含 ERROR 关键字的事件。"""
    for ev in events:
        if any("ERROR" in ln for ln in ev["lines"]):
            yield ev


if __name__ == "__main__":
    sample = [
        "INFO start",
        "INFO loading",
        "",
        "ERROR db failure",
        "INFO retry",
        "",
        "INFO shutdown",
    ]
    pipeline = keep_errors(enrich(read_events(iter(sample))))
    for e in pipeline:
        print(e)
    # {'id': 2, 'lines': ['ERROR db failure', 'INFO retry']}

这是”生成器管道”的经典案例——三个函数各司其职、惰性组合。

十四、常见陷阱

  1. 生成器只走一次：想反复用要 list()。
  2. yield 与 return 混用要小心：return 的值只出现在 StopIteration 里，普通调用者看不到。
  3. 生成器内部异常没处理：整个生成器就结束了，后面 yield 不会执行。
  4. 忘了消费：gen() 单纯调用不会执行内部代码——要 next(gen()) 或 for 循环才走。
  5. 两个生成器共享同一个数据源：迭代第一个会把源用掉，第二个空。
  6. 无限生成器忘 break/islice：程序卡死。

十五、小结与延伸阅读

  • yield 让函数变生成器，返回可迭代的生成器对象；
  • 生成器是惰性的——一次产出一个值；
  • 生成器表达式 (x for x in ...)；
  • yield from x 委托，扁平嵌套生成器；
  • 生成器可以 return 值，藏在 StopIteration.value；
  • 生成器 + 管道 = 高效数据流处理；
  • 生成器只走一次；
  • 现代协程用 async/await，不用 yield-based。

延伸阅读：

下一篇 装饰器进阶：带参数装饰器与类装饰器 我们把装饰器讲到更高阶。