Python 生成器从入门到精通:yield、迭代器协议与惰性求值

很多人写了几年 Python,对 yield 的认识还停留在"它能让函数返回多个值"。这个理解不算错,但太浅,浅到没法解释这些问题:为什么 yield 的函数调用后不执行?为什么生成器只能遍历一次?生成器和迭代器到底什么关系?yield from 又是什么?这篇文章把生成器、迭代器协议、惰性求值这三件事一次讲透,并给出能直接用在生产代码里的模式。

从迭代器协议说起

要理解生成器,得先理解 Python 的迭代器协议。当你写 for x in obj 时,Python 在背后做了这些事:

obj = [1, 2, 3]

# for 循环本质上等价于:
it = iter(obj)            # 1. 调用 iter() 拿到迭代器
while True:
    try:
        x = next(it)      # 2. 反复调用 next() 取值
    except StopIteration: # 3. 直到抛出 StopIteration 就停止
        break
    print(x)

所以"可迭代对象"(iterable)和"迭代器"(iterator)是两个不同的概念:

• 可迭代对象:实现了 __iter__ 方法,调用它能返回一个迭代器。list、dict、str、set 都是。
• 迭代器:实现了 __next__ 方法(同时也实现 __iter__ 返回自身)。它知道"下一个值是什么"以及"什么时候结束"。

手写一个迭代器,你就能彻底看清这个协议:

class CountDown:
    def __init__(self, start):
        self.current = start

    def __iter__(self):
        return self                 # 迭代器的 __iter__ 返回自己

    def __next__(self):
        if self.current <= 0:
            raise StopIteration     # 结束信号
        self.current -= 1
        return self.current + 1

for n in CountDown(3):
    print(n)                        # 3 2 1

能跑,但写起来很啰嗦:要维护 current 状态,要手动判断结束,要记得抛 StopIteration。生成器存在的意义,就是把这一整套样板代码自动化。

生成器:语法糖背后的状态机

把上面的 CountDown 用生成器重写,一下子缩短到三行:

def count_down(start):
    while start > 0:
        yield start
        start -= 1

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

关键认知:只要函数体里出现了 yield,这个函数就不再是普通函数,而是"生成器函数"。调用它不会执行任何代码,而是立刻返回一个生成器对象。

def gen():
    print('开始执行')
    yield 1
    print('继续执行')
    yield 2
    print('执行结束')

g = gen()                # 什么都不打印!只是创建了生成器对象
print(type(g))           # <class 'generator'>

print(next(g))           # 打印"开始执行",返回 1
print(next(g))           # 打印"继续执行",返回 2
print(next(g))           # 打印"执行结束",然后抛 StopIteration

每次调用 next(),函数从上次 yield 暂停的地方恢复执行,一直跑到下一个 yield 又暂停。函数的全部局部状态(变量值、执行到哪一行)都被引擎保存着。这就是生成器的本质 —— 它是一个可暂停、可恢复的函数,或者说一个由编译器自动生成的状态机。

为什么生成器只能遍历一次

因为生成器本身就是迭代器,它是"一次性"的 —— 内部状态走到尽头就回不去了。

g = count_down(3)
print(list(g))           # [3, 2, 1]
print(list(g))           # [] —— 已经耗尽,再遍历什么都没有

如果需要重复遍历,要么每次重新调用生成器函数拿一个新生成器,要么把结果固化成 list。这也是生成器和 list 最大的使用区别。

惰性求值:生成器真正的价值

生成器最大的优势不是"语法短",而是惰性求值(lazy evaluation):值只在被需要的那一刻才计算,不需要就不算,算过就丢。这带来两个直接好处 —— 省内存、能处理无限序列。

省内存:处理大文件

# 坏写法:一次性把整个文件读进内存
def read_all(path):
    with open(path) as f:
        return f.readlines()        # 10GB 文件直接 OOM

# 好写法:生成器逐行产出,内存占用恒定
def read_lines(path):
    with open(path) as f:
        for line in f:
            yield line.rstrip('\n')

# 配合生成器表达式做流式处理,全程不落地
def count_errors(path):
    lines = read_lines(path)
    errors = (l for l in lines if 'ERROR' in l)
    return sum(1 for _ in errors)   # 哪怕文件 100GB 也只占几 KB 内存

无限序列

def naturals():
    n = 0
    while True:                     # 无限循环也没关系,惰性的
        yield n
        n += 1

def take(gen, k):
    result = []
    for i, v in enumerate(gen):
        if i >= k:
            break
        result.append(v)
    return result

print(take(naturals(), 5))          # [0, 1, 2, 3, 4]

# 斐波那契无限流
def fib():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

print(take(fib(), 10))              # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

生成器表达式 vs 列表推导

nums = range(1, 1000001)

squares_list = [x * x for x in nums]   # 立刻算完,占约 8MB 内存
squares_gen  = (x * x for x in nums)   # 几乎不占内存,用一个算一个

# 只取前 3 个时,生成器的优势就出来了:它只算了 3 次
from itertools import islice
print(list(islice(squares_gen, 3)))    # [1, 4, 9]

经验法则:结果要反复用、要随机访问、要知道长度,用 list;结果只遍历一次、数据量大或无限、想做流式管道,用生成器。

yield from:生成器的委托

当一个生成器需要把另一个可迭代对象的值"原样转发"出去时,yield from 比手写循环更清晰,而且能正确传递 send、异常和返回值。

# 手写转发
def chain_manual(*iterables):
    for it in iterables:
        for item in it:
            yield item

# 用 yield from,等价但更清晰
def chain(*iterables):
    for it in iterables:
        yield from it

print(list(chain([1, 2], (3, 4), 'ab')))  # [1, 2, 3, 4, 'a', 'b']

# 递归遍历嵌套结构时,yield from 尤其优雅
def flatten(nested):
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)        # 递归委托
        else:
            yield item

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

生成器的双向通信:send、throw、close

生成器不只是"产出"值,还能"接收"值。yield 是一个表达式,它的值由外部 send() 进来。这让生成器可以充当协程。

def averager():
    total = 0.0
    count = 0
    average = None
    while True:
        value = yield average       # yield 既产出 average,又接收 send 进来的值
        total += value
        count += 1
        average = total / count

avg = averager()
next(avg)                            # 必须先"预激活"生成器,跑到第一个 yield
print(avg.send(10))                  # 10.0
print(avg.send(20))                  # 15.0
print(avg.send(30))                  # 20.0
avg.close()                          # 关闭生成器,内部会抛 GeneratorExit

throw() 能从外部往生成器内部抛异常,close() 会触发 GeneratorExit,让生成器有机会做清理。理解了这套机制,你就理解了 asyncio 早期是怎么用生成器实现协程的 —— async/await 在语义上正是 yield from 的进化。

实战模式:用生成器搭数据处理管道

生成器最优雅的工程用法,是把一个复杂的数据处理流程拆成一串小生成器,像 Unix 管道一样串起来。每一段只负责一件事,内存占用恒定,而且天然可组合、可测试。

import re

def read_lines(path):
    with open(path, encoding='utf-8') as f:
        for line in f:
            yield line.rstrip('\n')

def grep(pattern, lines):
    regex = re.compile(pattern)
    for line in lines:
        if regex.search(line):
            yield line

def parse_status(lines):
    for line in lines:
        m = re.search(r'status=(\d+)', line)
        if m:
            yield int(m.group(1))

def only_errors(codes):
    for code in codes:
        if code >= 500:
            yield code

# 像搭积木一样把管道拼起来 —— 整个过程文件不落地
pipeline = only_errors(parse_status(grep('api', read_lines('access.log'))))
error_count = sum(1 for _ in pipeline)
print(f'5xx 错误数:{error_count}')

这种写法的好处:每个函数都极短、单一职责、可单独写单元测试;整条管道是惰性的,处理 GB 级日志也只占几 KB;想加一个过滤步骤,插一个生成器进去即可,不用动其他代码。

常见坑与排查

坑一:生成器函数里的 return。 在生成器里 return value 不是"返回值",而是结束迭代,并把 value 塞进 StopIteration.value。普通 for 循环拿不到这个值,只有 yield from 能接住它。

def gen():
    yield 1
    return 'done'
    yield 2          # 永远执行不到

g = gen()
try:
    while True:
        print(next(g))
except StopIteration as e:
    print('返回值:', e.value)   # 返回值: done

坑二:把生成器当 list 反复用。 前面说过,生成器一次性。如果一个函数返回生成器,调用方又遍历了两次,第二次会静默地得到空结果 —— 这种 bug 不报错,最难查。

坑三:在生成器里用 with 但提前 break。 如果遍历方提前 break,生成器可能不会执行到 with 的退出逻辑,直到被 GC。需要确定性清理时,显式调用生成器的 close(),或用 contextlib.closing。

坑四:异常被吞。 生成器内部抛出的异常,只有在 next() 被调用到那一步时才会浮现。如果你创建了生成器却没遍历它,里面的错误代码根本不会运行 —— 这也是"调用生成器函数不报错,遍历时才报错"的原因。

itertools:生成器的标准库武器库

Python 的 itertools 模块是一整套基于生成器的工具,全部惰性、内存友好。手写生成器之前,先看看标准库是不是已经给好了。

from itertools import (
    count, cycle, repeat,           # 无限迭代器
    chain, islice, takewhile, dropwhile,  # 截取与拼接
    groupby, accumulate, pairwise,  # 聚合
)

# count:无限计数,可指定起点和步长
for i in islice(count(10, 2), 5):
    print(i)                         # 10 12 14 16 18

# chain:把多个可迭代对象首尾相连,不创建中间列表
list(chain([1, 2], [3, 4], [5]))     # [1, 2, 3, 4, 5]

# takewhile / dropwhile:按条件截断
list(takewhile(lambda x: x < 5, [1, 3, 5, 1]))  # [1, 3] —— 遇到 5 就停
list(dropwhile(lambda x: x < 5, [1, 3, 5, 1]))  # [5, 1] —— 丢到 5 之前

# accumulate:累积计算,默认累加
list(accumulate([1, 2, 3, 4]))       # [1, 3, 6, 10]

# groupby:按 key 分组(注意:只对相邻元素分组,通常先排序)
data = [('a', 1), ('a', 2), ('b', 3)]
for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))          # a [('a',1),('a',2)] / b [('b',3)]

# pairwise(3.10+):相邻两两配对,做差分、求斜率时很顺手
list(pairwise([1, 4, 9, 16]))        # [(1,4), (4,9), (9,16)]

这些工具的共同点是组合性 —— 它们的输入和输出都是可迭代对象,可以像管道一样层层嵌套,而整个链条始终是惰性的。

从生成器到协程:asyncio 的前身

前面讲过 send() 能让生成器接收外部传入的值。在 Python 3.4 的早期 asyncio 里,协程就是用"生成器 + yield from"实现的 —— 一个协程在等待 I/O 时 yield 出去,把控制权交还给事件循环,事件循环再在 I/O 就绪时 send() 唤醒它。

# Python 3.4 风格的协程(现在不推荐,但能看清原理)
import asyncio

@asyncio.coroutine
def old_style():
    yield from asyncio.sleep(1)       # 把控制权让给事件循环
    return 'done'

# Python 3.5+ 的现代写法 —— async/await 本质是上面的语法糖
async def modern():
    await asyncio.sleep(1)            # await 对应 yield from
    return 'done'

所以 async def 不是凭空出现的新东西:async 标记一个函数是协程,await 在语义上对应 yield from —— 都是"在这里暂停,把控制权交出去,等结果回来再恢复"。理解了生成器的"可暂停可恢复",再理解 async/await 就是顺理成章的事。两者的区别只在于:生成器面向"产出一串值"的场景,协程面向"等待一件异步事情完成"的场景,但底层的状态机机制是同一套。

性能对比:生成器到底省多少

import sys

# 列表:所有元素都在内存里
lst = [x for x in range(1000000)]
print(sys.getsizeof(lst))            # 约 8000000+ 字节(8MB 量级)

# 生成器:只有一个状态机对象
gen = (x for x in range(1000000))
print(sys.getsizeof(gen))            # 约 200 字节,与数据量无关

# 但要注意:生成器用 in 判断成员要从头扫,且会消耗它
print(500 in gen)                    # True,但此后 gen 已经走到了 500
print(100 in gen)                    # False!100 在 500 之前,已经被跳过了

最后一个例子是个隐蔽的坑:对生成器用 in 会消耗它。这再次提醒:生成器是一次性的流,不是可以随便回看的容器。把"它到底是流还是容器"这个问题想清楚,你就不会再写出"遍历两次发现第二次是空的"这类 bug。

写在最后

把这篇的核心收拢成一条链:迭代器协议(iter / next / StopIteration)是地基,生成器是编译器帮你自动实现这套协议的语法糖,它的本质是可暂停恢复的状态机,而它真正的价值在于惰性求值 —— 省内存、能处理无限流、能搭流式管道。

下次再写"先把数据全读进 list 再处理"的代码时,停一秒想想:这个 list 是不是只遍历一次?数据量会不会涨?如果是,把它换成生成器,几乎是免费的性能和内存优化。生成器不是高级技巧,它是 Python 处理"序列"这件事的默认姿势。

—— 别看了 · 2026