想象一下将 10 GB 日志文件加载到列表中只是为了查找错误。您的程序可能在找到内存之前就耗尽了内存。 Python 生成器 通过按需生成值（一次一个）来解决这个问题，这样您就可以在几乎没有内存的情况下处理大量甚至无穷无尽的数据。

本指南逐步展示了发电机运行时实际发生的情况。我们在 Python 3.13 上运行每个示例，并且输出来自该运行。它有助于熟悉函数和循环，我们的 Python 初学者教程 对此进行了介绍。

什么是发电机？

生成器是一种特殊类型的迭代器，您可以通过使用 yield 编写函数来创建它。任何带有 yield 的函数都会自动成为生成器函数，调用它会为您提供一个生成器对象，而不是运行代码。

def count_up(limit):
    print("starting")
    n = 1
    while n <= limit:
        yield n
        n += 1
    print("finished")

gen = count_up(2)
print(gen)

# <generator object count_up at 0x...>

请注意，“starting”没有打印出来。调用该函数仅创建了生成器。里面还没有运行任何东西。

产量如何运作，一步一步

每次调用 next() 都会运行该函数，直到到达 yield，交回该值并冻结到位。下一个调用将在该行之后立即恢复，并且每个局部变量都完好无损。

print(next(gen))
# starting
# 1

print(next(gen))
# 2

next(gen)
# finished
# StopIteration

仔细遵循输出。 “starting”仅出现在第一个 next() 上。在第二个，执行从停止的地方继续，因此 n 仍然可用并变为 2。在第三个，循环结束，“finished”打印，函数从末尾脱落，Python 引发 StopIteration 表示没有剩下任何东西。

您很少自己调用 next()。 for 循环会为您完成此操作并安静地处理 StopIteration，list()、sum() 和类似函数也是如此：

print(list(count_up(3)))

# starting
# finished
# [1, 2, 3]

收益率与回报

这种延迟被称为惰性求值：仅在需要时才计算值。生成器内的普通 return 只是结束它，从而触发 StopIteration。

测量的内存差异

我们使用 sys.getsizeof 将一百万个方格的列表与等效生成器进行了比较：

import sys

squares_list = [n * n for n in range(1_000_000)]
squares_gen = (n * n for n in range(1_000_000))

print(sys.getsizeof(squares_list), sys.getsizeof(squares_gen))

# 8448728 200

列表本身的结构占用了大约 8.4 MB，而生成器在我们的系统上只占用了 200 字节。真正的差距更大，因为 getsizeof 没有计算列表内的整数对象。对于十亿个项目，生成器的大小也保持不变，因为它只存储其当前状态，而不存储其结果。

不过，这需要权衡。列表可以通过 len() 进行重用、索引和测量。生成器不提供这些。如果您需要随机访问，或者需要对少量数据进行多次循环，则列表会更简单。有关这些类型的更多信息，请参阅我们的 Python 列表、元组与集合指南。

生成器表达式

对于简单的情况，您不需要函数。生成器表达式看起来像 列表理解，用圆括号代替方括号：

squares = [n * n for n in range(5)]   # builds a list now
squares = (n * n for n in range(5))   # produces values on demand

print(sum(n * n for n in range(5)))

# 30

当生成器表达式是函数的唯一参数时（如 sum() 中所示），您可以省略多余的括号。对于大数据的总计来说，这是一个整洁的习惯，因为不会创建临时列表。

构建数据管道

这就是发电机真正发挥作用的地方。小型生成器可以链接起来，以便每个项目一次一个地流经每一步：

def read_lines(lines):
    for line in lines:
        yield line.strip()

def only_errors(lines):
    for line in lines:
        if line.startswith("ERROR"):
            yield line

def messages(lines):
    for line in lines:
        yield line.split(" ", 1)[1]

log = ["INFO started\n", "ERROR db down\n",
       "ERROR disk full\n", "INFO stopped\n"]

print(list(messages(only_errors(read_lines(log)))))

# ['db down', 'disk full']

每个步骤只完成一项小工作，并且可以单独进行测试。在实际使用中，请将示例列表交换为打开的文件，因为文件对象可以逐行读取，如我们的 Python 文件处理指南 所示。即使是多千兆字节的日志也可以通过很少的内存。

无限序列

列表不可能是无限的，但生成器可以，因为它只计算下一个值。使用 itertools.islice 获取所需数量：

import itertools

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

print(list(itertools.islice(fibonacci(), 8)))

# [0, 1, 1, 2, 3, 5, 8, 13]

不过要小心：无限制地调用 list(fibonacci()) 会一直运行，直到你的内存耗尽为止。相同的模式适用于 ID 生成器、轮询循环和分页 API 结果，仅当当前页面用完时才获取下一页。

产量和批次

yield from 交给另一个可迭代对象，为您节省一个嵌套循环：

def chain_all(*groups):
    for group in groups:
        yield from group

print(list(chain_all([1, 2], (3, 4))))

# [1, 2, 3, 4]

一种非常实用的模式是将数据拆分为固定大小的批次，例如一次向数据库中插入一千行：

def batches(items, size):
    batch = []
    for item in items:
        batch.append(item)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch

print(list(batches(range(7), 3)))

# [[0, 1, 2], [3, 4, 5], [6]]

最后的if batch确保不会丢失较小的剩余组。 Python 3.12 及更高版本还包括 itertools.batched，它执行类似的工作。

发送、关闭和清理

生成器也可以接收值。 send() 恢复生成器，并且该值成为 yield 表达式的结果。您必须首先调用 next() 一次才能到达第一个 yield：

def running_total():
    total = 0
    while True:
        value = yield total
        total += value

t = running_total()
next(t)
print(t.send(10), t.send(5))

# 10 15

大多数人很少需要这个。更常见的问题是清理。如果提前停止，当生成器关闭时 finally 块仍然运行：

def tracked():
    try:
        yield 1
        yield 2
    finally:
        print("cleanup ran")

tr = tracked()
next(tr)
tr.close()

# cleanup ran

Pytest 使用了这个想法：带有 yield 的固定装置随后运行其清理，如我们的 pytest 教程 所示。同样，标准库的 contextmanager 装饰器将生成器变成 with 块。

类中的生成器

将 __iter__ 方法编写为生成器是使您的 自己的类 可迭代的最简单方法。一个好处是每个循环都会获得一个新的生成器，因此该对象可以一次又一次地循环：

class Countdown:
    def __init__(self, start):
        self.start = start

    def __iter__(self):
        n = self.start
        while n > 0:
            yield n
            n -= 1

c = Countdown(3)
print(list(c), list(c))

# [3, 2, 1] [3, 2, 1]

何时使用发电机

总体而言，生成器并不总是更快。如果您最终使用每个值，则总工作是相似的，并且对于小数据，列表理解甚至可以更快，因为它的每项开销更少。主要好处是内存使用，以及更快地获得第一个结果。

常见错误

重复利用耗尽的发电机

几乎每个人都会在某个时刻遇到这种情况。生成器会记住它已完成：

g = count_up(2)
print(list(g), list(g))

# [1, 2] []

第二次调用得到一个空列表，没有任何错误。再次调用该函数以获取新的生成器，或使用上面所示的类方法。

期望代码立即运行

def noisy():
    print("body runs")
    yield 1

n = noisy()        # nothing is printed
print(next(n))     # body runs, then 1

生成器顶部的参数检查直到第一个 next() 才会运行，因此错误可能会出现在远离导致错误的调用处。

其他陷阱

• 调用len()或索引。生成器都不支持，并提出TypeError。
• 过早关闭文件。 从 with 块内部返回生成器可能意味着文件在读取任何行之前关闭。将 yield 放入 with 块内。
• 将所有内容包装在 list(). 中，这会消除内存优势。
• 对微小数据使用一个。简单的列表通常更清晰。

最后的话

生成器是一个可以暂停的函数。这一想法为您提供了低内存使用、无尽的序列和干净的管道。为了将其付诸实践，请在代码中找到一个位置，该位置构建一个大列表，仅对其进行一次循环，然后将其转换为生成器。然后尝试在一个真实的日志文件上将两个小型生成器链接在一起。

常见问题

继续学习