Generator 和 yield 在 Python 中：处理大数据而不消耗 RAM

1. 总览

当处理数 GB 的日志文件或从数据库流式传输数百万条记录时，“读取所有内容到列表然后处理”的写法会导致程序消耗大量 RAM 并可能崩溃。本文帮助您理解 generator 和 yield 如何工作，与常规列表的对比，以及应用于实际问题（如读取大文件、数据处理管道）的方法。读完后，您将能够正确编写 generator，并知道何时不应该使用它。

2. 传统方法的问题

假设您需要读取一个 5GB 的日志文件，并计算包含“ERROR”的行数：

def count_errors_bad(filepath: str) -> int:
    with open(filepath, "r") as f:
        lines = f.readlines()  # 将整个文件加载到 RAM
    return sum(1 for line in lines if "ERROR" in line)

f.readlines() 将整个文件内容读取为内存中的字符串列表。对于 5GB 的文件，Python 会分配大约 5GB+ 的 RAM（加上对象开销）才开始计数。如果机器只有 4GB RAM，程序将收到 MemoryError 或系统因交换而冻结。

核心问题：您只需要一次处理一行，但代码却强制机器同时保持所有行。

3. Generator 是什么？

Generator 是一种特殊的 iterable：它按需生成值（lazy evaluation）而不是提前计算和存储所有内容。不是返回完整的列表，generator 返回一个一个的值，只在被请求时计算下一个值。

def count_errors_good(filepath: str) -> int:
    with open(filepath, "r") as f:
        return sum(1 for line in f if "ERROR" in line)

这里 f（文件对象）本身就是一个 iterator——当循环请求时它一次读取一行，不会将整个文件加载到 RAM。内存使用只是一次一行，无论文件是 5GB 还是 50GB。

4. yield 关键字：创建 generator 函数的方法

您可以通过在函数中使用 yield 来编写自己的 generator。当函数包含 yield 时，Python 自动将其转换为 generator 函数——调用它不会立即运行代码，而是返回一个 generator 对象。

def read_large_file(filepath: str):
    with open(filepath, "r") as f:
        for line in f:
            yield line.strip()

• yield line.strip() 在此处暂停函数，向外部返回值 line.strip()。
• 下一次调用 next() 时，函数从停止的地方继续运行，而不是从头开始。
• 函数的状态（局部变量、文件指针位置）在 yield 之间保持。

试试：

gen = read_large_file("access.log")
print(type(gen))
print(next(gen))
print(next(gen))

<class 'generator'>
127.0.0.1 - GET /index.html 200
127.0.0.1 - GET /style.css 200

注意：gen = read_large_file(...) 这一行不会打开文件，不读取任何内容。内部的代码只在您第一次调用 next(gen) 时运行。

5. 比较 RAM：generator vs list

要看清区别，可以使用 sys 模块来测量对象大小：

import sys

def squares_list(n: int) -> list[int]:
    return [i * i for i in range(n)]

def squares_gen(n: int):
    for i in range(n):
        yield i * i

big_list = squares_list(1_000_000)
big_gen = squares_gen(1_000_000)

print(f"List size: {sys.getsizeof(big_list):,} bytes")
print(f"Generator size: {sys.getsizeof(big_gen):,} bytes")

List size: 8,448,728 bytes
Generator size: 208 bytes

包含 100 万个元素的列表占用近 8.5MB，而 generator 只是一个小对象 208 字节——因为它不存储任何值，只存储状态以在需要时计算下一个值。

注意： Generator 只能遍历一次。遍历完后，再调用 next() 会抛出 StopIteration。如果需要多次遍历，必须通过再次调用函数来创建新的 generator。

6. Generator 表达式：更简洁的语法

类似列表推导式但使用圆括号而不是方括号，您可以使用 generator 表达式——无需定义单独的函数：

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

even_squares_gen = (n * n for n in numbers if n % 2 == 0)
print(sum(even_squares_gen))

220

使用 () 而不是 [] 使 Python 创建 generator 而不是立即创建列表。这在您只需要使用一次值的情况下很方便，例如直接传递给 sum()、max() 或 any()。

7. 使用 generator 构建数据处理管道

Generator 的强大之处在于可以链接多个处理步骤而不创建任何中间列表。假设您需要：读取订单 CSV 文件 → 筛选状态为“PAID”的订单 → 计算总金额。

import csv

def read_orders(filepath: str):
    with open(filepath, newline="") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

def filter_paid(orders):
    for order in orders:
        if order["status"] == "PAID":
            yield order

def extract_amount(orders):
    for order in orders:
        yield float(order["amount"])

def total_paid_amount(filepath: str) -> float:
    orders = read_orders(filepath)
    paid_orders = filter_paid(orders)
    amounts = extract_amount(paid_orders)
    return sum(amounts)

每个函数 read_orders、filter_paid、extract_amount 都是 generator——它们不立即处理全部数据，而是连接成一个管道。当 sum(amounts) 调用 next() 时，值通过每一步被拉取：读取 1 行 CSV → 检查状态 → 转换为浮点数 → 累加。在任何时刻，内存中只存在一条记录，无论 CSV 文件有 10 行还是 1000 万行。

7.1. 使用 pytest 测试管道

import pytest
from pathlib import Path

@pytest.fixture
def orders_csv(tmp_path: Path) -> str:
    content = (
        "order_id,amount,status\n"
        "1,100.0,PAID\n"
        "2,50.0,PENDING\n"
        "3,200.0,PAID\n"
    )
    file_path = tmp_path / "orders.csv"
    file_path.write_text(content)
    return str(file_path)

def test_total_paid_amount(orders_csv):
    result = total_paid_amount(orders_csv)
    assert result == pytest.approx(300.0)

tmp_path 是 pytest 内置的 fixture，为每个测试创建单独的临时目录。此测试创建一个包含 3 个订单的假 CSV 文件，其中只有 2 个是 PAID，并验证总计正好是 300.0。

8. yield from：委托给另一个 generator

当一个 generator 需要调用和嵌套另一个 generator 的结果时，yield from 比手动编写循环使代码更简洁。

def read_file_lines(filepath: str):
    with open(filepath) as f:
        for line in f:
            yield line.strip()

def read_multiple_files(filepaths: list[str]):
    for filepath in filepaths:
        yield from read_file_lines(filepath)

yield from read_file_lines(filepath) 等同于编写 for line in read_file_lines(filepath): yield line，但更简洁，并正确处理子 generator 具有 return 值的情况。read_multiple_files 函数允许您遍历多个日志文件，如同它们是一个连续的流，无需合并列表。

9. 使用 send() 将值发送到 generator

Generator 不仅生成向外的值——您还可以通过 send() 将值发送到其中，使其成为简单的 coroutine。

def running_average():
    total = 0.0
    count = 0
    average = None
    while True:
        value = yield average
        total += value
        count += 1
        average = total / count

avg_gen = running_average()
next(avg_gen)              # 启动 generator，运行到第一个 yield
print(avg_gen.send(10))    # 发送 10，average = 10/1
print(avg_gen.send(20))    # 发送 20，average = 30/2
print(avg_gen.send(30))    # 发送 30，average = 60/3

10.0
15.0
20.0

value = yield average 做两件事：向外返回 average，同时等待通过 send() 接收值分配给 value。您必须先调用一次 next(avg_gen) 来将 generator 推进到第一个 yield 点——如果在 generator 甚至没开始运行时就立即调用 send()，Python 会抛出 TypeError。

注意： 使用 send() 使代码比普通 generator 更难读。实际上，asyncio 等框架已经抽象化了这一机制；除非编写自己的库，否则您很少需要自己编写 send()。

10. 何时不应该使用 generator

Generator 并非总是比列表更好：

• 需要随机访问或多次遍历：generator 不支持 gen[5] 或从头再遍历。如果需要 len()、切片或多次遍历数据，列表更合适。
• 数据很小，不必担心 RAM：对于数千个元素，generator 的状态管理开销有时可能比列表推导式更慢，因为列表在 C 层级得到了优化。
• 需要并行处理或分配给多个 worker：列表易于分块以发送给 multiprocessing；generator 必须先“具体化”一部分才能分割。
• 调试更困难：打印列表可以立即看到所有值；打印 generator 只显示 <generator object ... at 0x...>，要查看值必须遍历（遍历后 generator“耗尽”，无法重用）。

如果不确定，简单原则是：当数据适应 RAM 且需要多次使用时使用列表；当数据很大、只需顺序处理一次或构建流式处理管道时使用 generator。

11. 结论

Generator 和 yield 准确解决了一个问题：处理大数据而无需将所有内容加载到 RAM，感谢 lazy evaluation 机制——只在被请求时计算值。您已经看到如何编写 generator 函数、generator 表达式、使用 yield from 将多个 generator 连接成管道，甚至通过 send() 将数据发送回 generator。记住这一权衡：generator 节省内存但代价是丧失重新遍历的能力、随机访问，有时代码可读性降低。为正确的问题选择正确的工具，而不是看到“节省 RAM”就在任何地方都使用 generator。