生成器与迭代器

What — 是什么

迭代器（Iterator）是实现了 __iter__ 和 __next__ 方法的对象，用于逐个访问集合中的元素。生成器（Generator）是创建迭代器的简洁方式——使用 yield 语句的函数自动成为生成器函数，调用时返回生成器对象。生成器实现了惰性求值（Lazy Evaluation），只在需要时计算下一个值。

核心概念：

  • 可迭代对象（Iterable）：实现了 __iter__ 方法的对象，返回迭代器
  • 迭代器（Iterator）：实现了 __next__ 方法的对象，每次返回一个值，耗尽后抛出 StopIteration
  • 生成器（Generator）：包含 yield 的函数自动变为生成器函数，调用返回生成器对象
  • 生成器表达式：(x*x for x in range(10))，列表推导式的惰性版本

关键特性：

  • 生成器函数遇到 yield 暂停执行，保留局部变量状态，下次从暂停处继续
  • 生成器是单向数据流，yield from 委托子生成器实现双向通道
  • 惰性求值：只在迭代时计算，不预先生成全部数据
  • 生成器只能迭代一次，耗尽后再次迭代返回空

运行机制：

  • 内存模型：生成器对象存储函数栈帧的快照（局部变量、执行位置），内存占用极小
  • 执行模型：next(gen) 恢复栈帧执行到下一个 yield，send(value) 将值注入 yield 表达式
  • 关闭机制：gen.close() 在 yield 处抛出 GeneratorExit，gen.throw(exc) 在 yield 处抛出指定异常
  • 委托机制：yield from subgen 将 send/throw/close 委托给子生成器，子生成器的返回值作为 yield from 表达式的值

迭代协议层次：

可迭代协议 (__iter__)  →  迭代器协议 (__iter__ + __next__)  →  生成器 (yield)
     ↑                            ↑                                  ↑
  for x in obj              next(it)                       惰性、可暂停
  list/dict/str             手动实现                       自动实现

Why — 为什么

适用场景：

  • 处理大数据集（逐行读取大文件、流式处理）
  • 无限序列（斐波那契、素数生成）
  • 管道式数据处理（多步转换链）
  • 协程基础（async/await 底层即生成器）
  • 内存敏感场景（不预先生成全部结果）

生成器 vs 列表对比：

维度生成器列表
内存O(1)（只存当前状态）O(n)（存储全部元素）
首次结果立即可用需等全部计算完成
多次迭代不支持（耗尽后为空）支持
索引访问不支持支持 O(1)
适用数据量无限/海量有限/适中
调试较难（状态不可见）容易（全部数据可见）

优缺点：

  • ✅ 优点：
    • 极低内存占用，适合处理海量数据
    • 惰性计算，首元素延迟极低
    • 代码简洁（yield 代替手写迭代器类）
    • 可组合为管道
  • ❌ 缺点：
    • 只能迭代一次，不能回退
    • 不支持索引和切片
    • 调试困难（无法查看全部数据）
    • 异常处理复杂（生成器中异常导致迭代终止）

How — 怎么用

快速上手

# 生成器函数
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 使用
fib = fibonacci()
for i, val in zip(range(10), fib):
    print(val, end=" ")  # 0 1 1 2 3 5 8 13 21 34

# 生成器表达式（惰性版列表推导式）
squares = (x * x for x in range(1_000_000))
print(next(squares))  # 1
print(next(squares))  # 4

# 对比内存
import sys
lst = [x * x for x in range(1000000)]
gen = (x * x for x in range(1000000))
print(sys.getsizeof(lst))  # ~8.5MB
print(sys.getsizeof(gen))  # ~200B

代码示例1：生成器管道 — 数据流式处理

def read_lines(filepath):
    """逐行读取大文件"""
    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip()

def filter_comments(lines):
    """过滤注释行"""
    for line in lines:
        if not line.startswith('#') and line:
            yield line

def parse_csv(lines):
    """解析CSV行"""
    for line in lines:
        fields = line.split(',')
        yield {
            'name': fields[0].strip(),
            'age': int(fields[1].strip()),
            'city': fields[2].strip()
        }

def filter_adults(records):
    """过滤成年人"""
    for record in records:
        if record['age'] >= 18:
            yield record

# 管道组合
pipeline = filter_adults(
    parse_csv(
        filter_comments(
            read_lines('people.csv')
        )
    )
)

for person in pipeline:
    print(person)
# 无需一次性加载整个文件到内存

代码示例2：yield from 与 send — 协作式多任务

# yield from：委托子生成器
def flatten(nested):
    """递归展平嵌套序列"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)  # 委托给子生成器
        else:
            yield item

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


# send：向生成器注入值
def accumulator():
    total = 0
    while True:
        value = yield total  # yield 返回 total，send 注入 value
        if value is None:
            break
        total += value

gen = accumulator()
next(gen)           # 启动生成器，返回 0
print(gen.send(10))  # 10
print(gen.send(20))  # 30
print(gen.send(5))   # 35

# yield from 的返回值
def subgen():
    yield 1
    yield 2
    return "subgen done"

def main_gen():
    result = yield from subgen()  # result 接收子生成器的返回值
    yield f"main got: {result}"

print(list(main_gen()))  # [1, 2, 'main got: subgen done']

代码示例3：itertools 实用组合

from itertools import (
    chain, islice, groupby, count,
    cycle, repeat, combinations, permutations,
    product, accumulate, starmap, zip_longest,
    takewhile, dropwhile, filterfalse
)

# chain：合并多个可迭代对象
list(chain([1, 2], [3, 4], [5]))  # [1, 2, 3, 4, 5]

# islice：切片迭代器
list(islice(count(10), 5))  # [10, 11, 12, 13, 14]

# groupby：分组（需先排序）
data = sorted([('A', 1), ('B', 2), ('A', 3), ('B', 4)], key=lambda x: x[0])
for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# A [('A', 1), ('A', 3)]
# B [('B', 2), ('B', 4)]

# accumulate：累积计算
list(accumulate([1, 2, 3, 4, 5]))  # [1, 3, 6, 10, 15]
list(accumulate([1, 2, 3, 4], initial=0))  # [0, 1, 3, 6, 10]

# product：笛卡尔积
list(product('AB', [1, 2]))  # [('A',1), ('A',2), ('B',1), ('B',2)]

# zip_longest：不等长合并
list(zip_longest([1, 2, 3], ['a', 'b'], fillvalue='-'))  # [(1,'a'), (2,'b'), (3,'-')]

# takewhile / dropwhile
list(takewhile(lambda x: x < 5, [1, 3, 5, 2, 4]))  # [1, 3]
list(dropwhile(lambda x: x < 5, [1, 3, 5, 2, 4]))  # [5, 2, 4]

常见问题与踩坑

问题原因解决方案
生成器只能迭代一次迭代后内部状态耗尽需要多次迭代用 list(gen) 或重新创建
忘记 next() 启动生成器send 前必须先 next()用 gen.send(None) 或 next(gen) 启动
生成器中异常丢失异常传播后生成器关闭用 try/finally 确保清理，或 contextlib
yield from 不返回值子生成器没有 return子生成器 return value 后 yield from 可接收
大管道调试困难无法查看中间状态用 tee 复制流或插入日志生成器
内存泄漏生成器持有大对象的引用及时 close() 或用完即弃

最佳实践

  • 处理大数据用生成器管道，避免一次性加载到内存
  • 用 itertools 代替手写循环，性能更好且更 Pythonic
  • 生成器中用 try/finally 确保资源释放
  • 复杂生成器逻辑用 yield from 委托，保持扁平
  • contextlib 提供了生成器版上下文管理器（@contextmanager）
  • 生成器表达式优于列表推导式——除非需要多次迭代或索引

面试题

Q1: 迭代器和可迭代对象有什么区别？

可迭代对象（Iterable）实现了 __iter__ 方法，返回一个迭代器。迭代器（Iterator）同时实现了 __iter__ 和 __next__，__iter__ 返回自身，__next__ 返回下一个值并在耗尽时抛出 StopIteration。可迭代对象可以被 for 循环遍历，但不能直接 next()；迭代器可以。iter(obj) 获取迭代器，next(it) 获取下一个值。

Q2: 生成器函数和普通函数有什么区别？

生成器函数包含 yield 语句，调用时不执行函数体，而是返回一个生成器对象。每次 next() 时执行到 yield 暂停，下次从暂停处继续。普通函数调用时立即执行，遇到 return 结束并返回值。生成器是惰性的、可暂停的，内存占用极低；普通函数是急切的、一次性的。

Q3: yield from 的作用是什么？它解决了什么问题？

yield from subgen() 将 send/throw/close 操作委托给子生成器，同时子生成器的返回值作为 yield from 表达式的值。它解决了：1) 手动 for 循环 yield 子生成器元素无法传递 send 值的问题；2) 简化嵌套生成器的委托逻辑；3) 子生成器异常的正确传播。yield from 是 asyncio 早期（Python 3.4）实现协程的基础。

Q4: 生成器的 send 方法是如何工作的？

gen.send(value) 恢复生成器执行，并将 value 作为当前 yield 表达式的值注入。首次必须 send(None) 或 next(gen) 启动生成器（因为此时还没有 yield 表达式等待值）。send 的典型应用是协程通信：生成器 yield 出中间结果，外部 send 注入新数据继续计算。

Q5: 如何让生成器支持 with 语句？

使用 contextlib.@contextmanager 装饰器：yield 之前的代码是 __enter__，yield 的值绑定到 as 变量，yield 之后的代码是 __exit__。异常在 yield 处抛出，可用 try/except 捕获处理。这是创建上下文管理器最简洁的方式，比手写 __enter__/__exit__ 类简单得多。

Q6: 生成器表达式的求值时机是什么？和列表推导式有什么区别？

生成器表达式 (x*x for x in range(10)) 创建时不计算任何值，只在迭代时逐个求值。列表推导式 [x*x for x in range(10)] 创建时立即计算全部值并存储。生成器表达式是惰性的，内存 O(1)；列表推导式是急切的，内存 O(n)。但生成器只能迭代一次，列表可反复使用。

Q7: itertools.groupby 的注意事项是什么？

groupby 要求输入已按分组键排序！它只合并相邻的相同键元素。如果未排序，相同键的元素会被分到不同组。这是最常见的误用。正确做法：先用 sorted(data, key=...) 排序，再用 groupby。另外，groupby 返回的分组迭代器只能消费一次，需及时处理。

Q8: 如何实现一个可重置的生成器？

生成器本身不可重置，但可以通过以下方式模拟：1) 将生成器逻辑封装为类，实现 __iter__ 每次返回新的生成器；2) 用函数工厂（def make_gen(): return (x for x in ...)），每次调用创建新实例；3) 用 itertools.tee 复制迭代器（但会缓存已消费的值，内存代价高）。最佳方案是方案 1——用类包装，__iter__ 每次返回新生成器。

相关链接：