Python生成器进阶 从yield到yield from实现省内存生产者消费者模式

Python 的生成器（Generator）是这门语言里最被低估的高级特性之一。许多开发者仅仅把它当作一个“懒加载”的迭代器来使用，但实际上，它内置了一整套双向通信和协程组合的能力。一旦掌握了 send()、throw()、close() 以及 yield from，在处理大数据流、实现协程、构建流水线模式时，代码不仅能写得优雅简洁，还能把内存占用压到极致。下面我们一步步拆解这些核心机制。

一、知识点简介

生成器是 Python 最被低估的高级特性之一。很多人只把它当作“惰性迭代器”来用，实际上 send()/throw()/close() 让生成器拥有了双向通信能力，yield from 则提供了生成器委托/协程组合的能力。掌握这些，你可以在处理大数据流、实现协程、构建管道模式时写出极其优雅且节省内存的代码。

二、核心概念速览

三、核心代码 & 使用示例

1. 基础：省内存的大文件逐行处理

def read_large_file(filepath: str):
    """逐行读取，任何时候只保留一行在内存"""
    with open(filepath, encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# 使用（100MB 文件也毫无压力）
for line in read_large_file("big_log.txt"):
    if "ERROR" in line:
        print(line)

2. 进阶：用 send() 实现双向通信 —— 简易累加器

def running_a verage():
    """动态累加器：外部边送数据边拿当前均值"""
    total, count = 0.0, 0
    a verage = None
    while True:
        value = yield a verage          # yield 产出上次均值，暂停等待 send
        total += value
        count += 1
        a verage = total / count

a vg = running_a verage()
next(a vg)                             # 启动生成器（走到第一个 yield）
print(a vg.send(10))   # 10.0
print(a vg.send(20))   # 15.0
print(a vg.send(30))   # 20.0
print(a vg.send(40))   # 25.0

3. 实战：生产者-消费者协程模式

def consumer():
    """消费者：不断接收数据并处理"""
    items = []
    while True:
        item = yield
        if item is None:          # 哨兵值，终止
            break
        items.append(item)
        print(f"[消费] {item}")
    return items

def producer(data, consumer_gen):
    """生产者：发送数据给消费者"""
    next(consumer_gen)            # 启动
    for item in data:
        consumer_gen.send(item)
    consumer_gen.send(None)       # 发送终止信号

# 运行
c = consumer()
producer(["A", "B", "C", "D"], c)
# 输出：
# [消费] A
# [消费] B
# [消费] C
# [消费] D

4. yield from —— 生成器委托，优雅组合

def sub_gen_a():
    yield "A1"
    yield "A2"

def sub_gen_b():
    yield "B1"
    yield "B2"

def main_gen():
    yield "START"
    yield from sub_gen_a()      # 委托：自动转发 send/throw/close
    yield "MIDDLE"
    yield from sub_gen_b()
    yield "END"

print(list(main_gen()))
# ['START', 'A1', 'A2', 'MIDDLE', 'B1', 'B2', 'END']

yield from 的真正威力：它会自动将 send() / throw() / close() 转发到子生成器，让外部代码可以直接与最内层生成器通信，无需手动传递。

5. 实战：流水线（Pipeline）模式处理数据流

def lines(filepath):
    with open(filepath) as f:
        yield from f

def filter_lines(lines_gen, keyword):
    for line in lines_gen:
        if keyword in line:
            yield line

def parse_json_lines(lines_gen):
    import json
    for line in lines_gen:
        yield json.loads(line)

def extract_field(json_gen, field):
    for obj in json_gen:
        yield obj.get(field)

# 组装管道（零拷贝，全程惰性求值）
pipe = extract_field(
    parse_json_lines(
        filter_lines(
            lines("data.jsonl"),
            "ERROR"
        )
    ),
    "timestamp"
)

for ts in pipe:
    print(f"错误时间: {ts}")

每条管道节点都是一个生成器，数据逐条流过，内存占用 = 一条记录的大小，与文件总大小无关。

6. 生成器表达式 vs 列表推导 —— 内存对比

import sys

list_comp = [x ** 2 for x in range(10_000_000)]   # 立即构建全部
gen_expr  = (x ** 2 for x in range(10_000_000))    # 惰性，不占内存

print(f"列表推导: {sys.getsizeof(list_comp) / 1024 / 1024:.1f} MB")  # ~305 MB
print(f"生成器:   {sys.getsizeof(gen_expr)} bytes")                   # ~112 字节

四、注意事项 / 避坑指南

1. 生成器只能遍历一次 —— 第二次遍历将返回空结果。如果需要复用，可以用 list() 转换为列表，或使用 itertools.tee() 复制。
2. 必须 next() 启动 —— 对于带 send() 的生成器，第一次 send() 必须传 None，否则会抛出 TypeError。
3. 异常处理 —— 生成器内部抛出的异常会向上冒泡到调用方，可以使用 gen.throw() 来测试生成器的异常处理能力。
4. 关闭生成器 —— 不再使用时可以调用 gen.close() 释放资源，生成器内部可以用 try/finally 做清理操作。
5. 不要混用 return 和 yield —— 从 Python 3.3+ 开始，生成器 return 的值会附加到 StopIteration.value 中，需通过 yield from 捕获。

def gen_with_return():
    yield 1
    yield 2
    return "完成"

g = gen_with_return()
print(list(g))  # [1, 2] —— return 值被忽略！

# 正确获取 return 值的方式
def wrapper():
    result = yield from gen_with_return()
    print(f"子生成器返回: {result}")

list(wrapper())  # 输出: 子生成器返回: 完成

五、总结

生成器远不止是“省内存的迭代器”：

• send() —— 让外部和生成器双向对话，实现协程模式
• yield from —— 优雅组合生成器，自动转发消息
• 流水线模式 —— 链式处理大数据流，每条记录在管道中逐级流过

下次面对大量数据处理或需要生产者-消费者模式的场景，用生成器代替先组装列表再处理的写法，不仅代码更优雅，内存性能也会显著提升。