生成器基础与核心原理

生成器是Python中实现惰性计算的典型范例，其本质是使用yield语句的特殊函数。与传统函数一次性返回所有结果不同，生成器通过yield暂停执行并保留上下文状态，在每次迭代时恢复执行直到下一个yield。

def timestamp_generator(start, end):
    current = start
    while current <= end:
        yield current
        current += 1

# 使用示例
gen = timestamp_generator(1743665549, 1743665549 + 10)
for ts in gen:
    print(f"Processing timestamp: {ts}")

关键实现原理：
1. 调用生成器函数时返回的是生成器迭代器对象
2. 每次调用next()时执行到下一个yield语句
3. 通过gi_frame属性保存的栈帧维持局部变量状态
4. 抛出StopIteration异常标识迭代终止

高级生成器特性与应用

协程与双向通信

Python 2.5+通过.send()方法实现了生成器与调用方的双向通信，这是协程实现的基础：

def coroutine_gen():
    received = yield "Ready"
    while received:
        received = yield f"Processed: {received.upper()}"

cg = coroutine_gen()
print(next(cg))  # 输出: Ready
print(cg.send("data"))  # 输出: Processed: DATA

yield from语法

Python 3.3引入的yield from简化了嵌套生成器的场景：

def sub_gen():
    yield from range(3)

def main_gen():
    yield from sub_gen()
    yield from (x*2 for x in sub_gen())

list(main_gen())  # 输出: [0, 1, 2, 0, 2, 4]

性能优化实践

内存效率对比

生成器在处理大规模数据时展现显著优势：

import sys

def regular_list(n):
    return [x**2 for x in range(n)]

def generator_version(n):
    return (x**2 for x in range(n))

# 内存占用测试
n = 1000000
print(sys.getsizeof(regular_list(n)))  # 约8.5MB
print(sys.getsizeof(generator_version(n)))  # 112字节

流水线处理模式

生成器链式调用形成高效处理流水线：

def read_logs(file):
    with open(file) as f:
        for line in f:
            yield line.strip()

def filter_errors(logs):
    for log in logs:
        if "ERROR" in log:
            yield log

def extract_timestamps(logs):
    for log in logs:
        ts = log.split()[0]
        yield int(ts)

# 组合使用
pipeline = extract_timestamps(filter_errors(read_logs("app.log")))
error_times = list(pipeline)

行业应用场景分析

大数据处理

• 优势：避免将全部数据加载到内存
• 限制：单次遍历特性导致无法重复使用
• 实践案例：Apache Beam等框架底层使用生成器模式处理数据流

异步编程基础

• 生成器协程：早期asyncio实现基础
• 现代替代：Python 3.5+的async/await语法更直观
• 迁移建议：新项目应优先使用原生协程

最佳实践与陷阱规避

推荐模式

1. 使用生成器表达式替代临时列表

   # 优于: sum([x*x for x in range(10)])
   sum(x*x for x in range(10))
   

2. 文档化生成器协议

   def batch_processor(items, size):
       """生成固定大小的数据批次
   
       Args:
           items: 可迭代输入数据
           size: 每批数量
   
       Yields:
           list: 包含最多size个元素的列表
       """
       batch = []
       for item in items:
           batch.append(item)
           if len(batch) == size:
               yield batch
               batch = []
       if batch:
           yield batch
   

常见陷阱

• 资源泄漏：生成器中打开的文件需确保关闭

  # 错误示范
  def read_files():
      for name in ['a.txt', 'b.txt']:
          yield open(name).read()
  
  # 正确做法
  def read_files_safe():
      for name in ['a.txt', 'b.txt']:
          with open(name) as f:
              yield f.read()
  

• 重复消费问题：生成器只能迭代一次

  gen = (x for x in range(3))
  list(gen)  # [0, 1, 2]
  list(gen)  # []
  

现代Python中的演进

Python 3.8+引入了海象运算符进一步优化生成器表达式：

import random

def random_data_stream():
    while (value := random.random()) > 0.1:
        yield value

# 获取首个大于0.8的值
first_large = next(x for x in random_data_stream() if x > 0.8)

标准库中的典型应用：
– itertools模块提供无限生成器（count/cycle等）
– contextlib.contextmanager实现生成器上下文管理器
– pathlib.Path.glob()返回生成器而非列表