问题与目标
读取大文件、分页结果或连续数据流时,一次性构造列表会浪费内存。迭代协议让程序逐项取值,生成器则用普通函数的写法实现惰性计算。
完成标准:能区分可迭代对象与迭代器,解释 iter()、next() 和 yield,并完成一个逐行处理大文件的程序。
核心概念
可迭代对象能通过 iter(obj) 得到迭代器;迭代器通过 next() 返回下一个值,结束时抛出 StopIteration。for 循环替开发者完成了这两步。
列表可重复遍历,但列表的迭代器通常是一次性的。生成器函数包含 yield,调用函数时不会立刻执行函数体,而是返回生成器;每次取值执行到下一个 yield 暂停。
惰性计算减少峰值内存,但也意味着错误可能在消费数据时才出现,而且数据通常不能无成本地重新遍历。
下面这个最小迭代器把协议显式写了出来:
class Countdown:
def __init__(self, start: int) -> None:
self.current = start
def __iter__(self) -> "Countdown":
return self
def __next__(self) -> int:
if self.current <= 0:
raise StopIteration
value = self.current
self.current -= 1
return value
counter = Countdown(3)
print(next(counter))
print(list(counter))
输出 3 和 [2, 1],也说明迭代器保存消费位置、不会自动回到起点。生成器把 __iter__、__next__ 和暂停状态交给解释器实现。
可运行实现
保存为 stream_errors.py:
from collections.abc import Iterator
from pathlib import Path
def error_lines(path: Path) -> Iterator[tuple[int, str]]:
with path.open(encoding="utf-8") as file:
for line_number, line in enumerate(file, start=1):
text = line.rstrip("\n")
if "ERROR" in text:
yield line_number, text
def main() -> None:
path = Path("app.log")
path.write_text(
"INFO service started\nERROR database timeout\nINFO retry\n",
encoding="utf-8",
)
for number, text in error_lines(path):
print(f"{number}: {text}")
if __name__ == "__main__":
main()
运行 python stream_errors.py,输出:
2: ERROR database timeout
函数输入是文件路径,输出不是列表,而是一个逐项产生 (行号, 内容) 的迭代器。无论文件多大,程序只需保留当前行附近的数据。
生成器表达式也保持惰性:
numbers = (number * number for number in range(1_000_000))
first_three = [next(numbers) for _ in range(3)]
print(first_three)
与列表推导式相比,它不会立刻生成一百万个结果。是否节省内存可用同一份输入分别构造列表和生成器,再通过 tracemalloc 或进程监控比较峰值,而不是仅凭语法判断。
常见问题与排查
- 生成器没有输出:只创建生成器不会执行,需要
for、next()或其他消费操作。 - 第二次遍历为空:同一生成器已经耗尽;需要重新调用生成器函数。
yield和return混淆:yield产生一个值并暂停,return结束生成器。- 在生成器外打开文件后立刻关闭:消费时文件已不可用。示例把
with放在生成器内部,使资源生命周期覆盖迭代过程。 - 想同时保存所有结果:可以
list(error_lines(path)),但会失去流式内存优势。
小结
迭代器描述“如何取下一个值”,生成器降低了实现迭代器的成本。面对大文件和流式结果,先问是否真的需要把全部数据放进内存。
License: CC BY-NC 4.0
Updated 3 hours ago
Was this article helpful? Give it a like.
0 comments


