Python迭代器与生成器：从内存优化到流式处理实践

先问一个问题：你在Python里处理过几个GB的文件吗？如果处理过，大概率经历过内存暴涨、程序卡死，最后不得不把代码改成一行一读。我前阵子帮朋友优化一个日志分析脚本，逻辑本身不复杂，就是用了data = f.read()，2GB的日志差点把16GB内存的机器跑死。那次之后我又翻了一遍迭代器和生成器的底层机制，发现很多性能问题其实都出在这两个特性没吃透上。

Python的迭代器与生成器，说起来就是“按需取值、惰性计算”这八个字。列表、字典、文件对象的for循环，底层全是迭代器协议在工作；而生成器作为迭代器的生产工厂，能让你用yield写出近乎魔法般的内存友好代码。这篇内容从迭代协议讲起，把yield的暂停机制、大文件流式处理、数据批量加载、无限序列处理，再到send/throw/close这些进阶用法完整过一遍。刚接触Python想提升代码性能的，或者写了一阵子但对这两个概念一知半解的，都可以从这篇里找到能直接用的东西。

1. 迭代器与可迭代对象：先搞清Python的迭代协议

1.1 三个容易混淆的概念：可迭代对象、迭代器、迭代

先把这个最基本的区分讲透，因为很多人把“可迭代对象”和“迭代器”当成一回事，结果写代码时被各种TypeError追着跑。

可迭代对象，指能用for循环遍历的对象，比如列表、字典、字符串、range对象。迭代器，指实现了__next__()方法、可以逐个吐值的对象，文件对象就是典型。而迭代，是for循环遍历这个动作本身。

判断方式很直接：

from collections.abc import Iterable, Iterator print(isinstance([1, 2, 3], Iterable)) # True print(isinstance([1, 2, 3], Iterator)) # False print(isinstance(iter([1, 2, 3]), Iterator)) # True

这里的关键是iter()函数。iter()会调用对象的__iter__()方法，拿到一个迭代器。比如列表本身不是迭代器，但iter([1, 2, 3])返回的list_iterator对象就是。迭代器还有一个特征：它自己也实现了__iter__()，只不过返回的是自己。所以你可以说迭代器一定是可迭代对象，但反过来不成立。

for循环本质上就是迭代器的重复调用。你用while手写一遍就明白了：

lst = [10, 20, 30] it = iter(lst) while True: try: value = next(it) print(value) except StopIteration: break

一个简单类比：可迭代对象像是货架，迭代器像是按顺序给你递货物的传送带。你问传送带要下一个，它给你一个；货架空了，它抛一个StopIteration表示“没了”。for循环就是那个一直“按一下按钮”取货的人。

1.2 动手写一个迭代器：MyRange的完整拆解

理解了协议之后，自己实现一个迭代器是必经之路。下面这个MyRange类，用__iter__和__next__模拟内置range的一部分行为：

class MyRange: def __init__(self, start, end): self.current = start self.end = end def __iter__(self): return self def __next__(self): if self.current >= self.end: raise StopIteration value = self.current self.current += 1 return value

用起来和range差不多：

for i in MyRange(1, 5): print(i) # 输出：1 2 3 4

这个类里有两个关键设计。一是__iter__返回self，表示“我既是可迭代对象，也是迭代器”，在for循环里能直接使用。二是__next__内部维护一个current状态，每次调用都更新状态，直到边界条件触发StopIteration。如果没有异常终止机制，for循环会无限跑下去。

手写类的痛点也很明显：维护状态太啰嗦。要是生成斐波那契数列、处理多层级嵌套，状态变量和边界条件会越来越多。这时候就该轮到生成器登场了。

1.3 迭代器的一次性陷阱：遍历完就没了

迭代器和列表最大的区别之一是一次性。列表可以来回遍历，迭代器取完就空。我早期写过一段代码，定义一个迭代器，第一个for循环正常输出，第二个for循环什么结果都没有，排查了半天才反应过来：同一个迭代器对象已经被第一次遍历耗尽了。

it = iter([1, 2, 3]) list(it) # [1, 2, 3] list(it) # []

另外list(iterator)会把迭代器里所有值全取出来，一旦执行完，迭代器也就废了。很多人在做数据流处理时，喜欢顺手list(generator)转列表，结果一轮用下来，内存没省，迭代器也没得用了。

解决方案有两个：需要多次遍历就一次性转成列表；不想占内存就重新创建迭代器。注意自定义迭代器类（比如MyRange）如果__iter__返回self，多次for循环同样会出问题，第二次循环拿到的是状态已经走到头的同一个对象。标准做法是让__iter__每次返回一个全新的迭代器，这也是后面生成器更省心的原因之一。

2. 生成器与yield：一次读懂惰性计算的执行机制

2.1 生成器函数：函数里出现yield就完全不一样了

任何函数，只要函数体里出现了yield关键字，它就不再是普通函数，而是生成器函数。调用它不会执行函数体，而是返回一个生成器对象。看代码：

def gen(): print("生成器函数开始执行") yield 1 print("第一次yield之后继续") yield 2 print("执行结束") g = gen() print(type(g)) # <class 'generator'> print("这个print会先输出")

当你调用gen()时，“生成器函数开始执行”根本不会打印。直到你调用next(g)，函数体才开始运行，遇到第一个yield 1，把1返回给调用方，然后暂停。再调next(g)，从暂停点继续，打印“第一次yield之后继续”，遇到yield 2，返回2，再次暂停。第三次next(g)，打印“执行结束”，函数走到末尾，自动抛出StopIteration。

这套执行流程，和普通函数“一次性从头跑到尾”完全不同。生成器像一部可以随时暂停、按遥控器接着放的电影，暂停点就是yield那一行，函数里的局部变量、循环进度、调用栈都原封不动保留着。

这样就可以完全替代手写迭代器类：

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

对比MyRange类，生成器函数不需要__iter__、__next__，不用维护self.current，更不用手动抛StopIteration。函数结束自动就抛了。这也是为什么实际项目里几乎没人手写迭代器类，全用生成器。

2.2 yield的底层状态机：挂起、保存、恢复

理解yield的关键，在于理解“状态保存”。生成器对象内部其实是一个状态机，保存了两样东西：当前执行到的指令位置和当前的局部变量表。每次next调用，从保存的位置继续执行，直到下一个yield或函数结束。

这跟书签是一个道理。你读一本书，看到第100页停下，书签夹在100页。下次拿起书，直接在100页接着看。生成器的“书签”就是yield表达式那一行，局部变量就是“你脑子里记住的剧情线”。

因为局部变量跨越了多次调用存活，生成器才能表现出“有记忆”的特性。比如这个无限计数器：

def count_from(start): n = start while True: yield n n += 1

n在每次yield之后都保留着旧值，下一次恢复递增后再yield。如果你用纯函数写这个逻辑，根本写不出来——普通函数调用结束，局部变量就销毁了。

也正因如此，生成器不支持索引和长度查询。len(generator)会抛TypeError，generator[0]也不行。它不是一个容器，而是一个状态流。想取第5个值？老老实实调用5次next，或者用islice，后面会讲。

2.3 生成器表达式：内存更省的列表推导式平替

生成器表达式和列表推导式长得几乎一样，只是方括号换成圆括号：

list_squares = [x * x for x in range(1000000)] gen_squares = (x * x for x in range(1000000))

列表推导式会立刻算出一百万个平方数放进列表，生成器表达式则是一个惰性计算的生成器对象。用sys.getsizeof看看差距：

import sys print(sys.getsizeof([x * x for x in range(1000000)])) # 几百万字节，随列表增长 print(sys.getsizeof((x * x for x in range(1000000)))) # 固定的一百多字节

百万级列表随时能到几MB甚至几十MB，生成器对象永远是那个固定大小的小对象。它不是“内存省了一点”，而是“无论数据多大，内存都不涨”。

但生成器表达式不是万能的。它只能遍历一次，不能重复使用，也没有len和索引。如果你只是想把一个大的计算结果传给某个需要二次遍历的函数，用生成器没问题；但如果你需要随机访问、多次遍历或者知道长度，老老实实转列表反而更好。

我个人的取舍标准很简单：数据量大、只遍历一次、可以一边生成一边消费，用生成器表达式；数据量小、要反复操作，用列表推导式。盲目追求生成器表达式，反而会把代码写得别扭。

3. 实操落地：大文件处理与数据批量加载方案

3.1 大文件流式处理：别再read()整个文件了

很多人读文件第一反应是data = f.read()，文件多大内存就占多大。我在开头提到的日志分析就是这样崩的。合理做法是让文件对象自己当迭代器，一行一行地读：

with open("huge.log", "r") as f: for line in f: process(line)

文件对象本身实现了迭代器协议，for循环每次从磁盘读取一行并返回，内存里最多只保留当前这一行。2GB的日志文件，处理时内存占用从GB级直接降到几MB级。

如果文件二进制处理需要按固定大小分块，可以写一个分块生成器：

def read_chunks(file_path, chunk_size=8192): with open(file_path, "rb") as f: while True: chunk = f.read(chunk_size) if not chunk: break yield chunk # 使用 for chunk in read_chunks("data.bin", 4096): process_chunk(chunk)

这里有个很容易踩的坑：按块读二进制文件没问题，但按块读文本文件可能把多字节字符截断。比如UTF-8编码的中文字符占3个字节，两块的分界线正好切在字符中间，解码时就报UnicodeDecodeError。处理文本场景，优先用按行读取；实在要按块，建议用TextIOWrapper配合buffer控制解码，或者不按字节大小切块，而是按“累积到指定行数再yield”的方式切块。

3.2 数据批量加载：batch迭代器的三种写法

批量处理是生成器最实用的场景之一，比如数据库批量插入、深度学习训练数据batch、大批量写Excel。最容易想到的方式是切片：

for i in range(0, len(data), batch_size): batch = data[i:i + batch_size] process(batch)

但这种方式要求data支持切片，而且是一次性加载到内存的。换成batch生成器，就能处理任意可迭代对象，包括流数据：

def batch_iter(iterable, batch_size): batch = [] for item in iterable: batch.append(item) if len(batch) == batch_size: yield batch batch = [] if batch: yield batch # 用法 data_range = range(10) for batch in batch_iter(data_range, 3): print(batch)

输出结果：

[0, 1, 2] [3, 4, 5] [6, 7, 8] [9]

这个生成器不管传入的是列表、文件迭代器还是另一个生成器，都能按batch吐数据。配合yield from还能做级联展开，比如处理二维嵌套结构：

def flatten(nested): for sublist in nested: yield from sublist pairs = [[1, 2], [3, 4], [5, 6]] print(list(flatten(pairs))) # [1, 2, 3, 4, 5, 6]

注意一点：如果训练任务需要对batch内的数据shuffle，直接对当前batch洗牌没问题；但千万别为了“先shuffle整个数据集再分批”，把传入batch_iter的生成器转成列表掏空内存。正确做法是分批洗牌，或者用支持随机访问的列表场景单独处理。

3.3 无限序列与数据流：islice正确消费生成器

生成器能表达无限序列，这是普通列表做不到的。比如斐波那契数列：

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

直接list(fib())会把内存耗尽，因为根本没有终点。但配合itertools.islice，就能安全地取前N个：

from itertools import islice print(list(islice(fib(), 10))) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

islice(generator, n)的意思是从生成器里取前n个值，取出后生成器还在原位置。这个方法在处理无限流的场景尤为重要：传感器数据流、不中断的任务队列、持续产生的日志流。不要试图把无限生成器转成列表，要用“限量消费”的方式按需取用。

itertools里还有不少能配合生成器使用的工具。chain把多个生成器串成一条流水线：

from itertools import chain for item in chain(range(3), "abc"): print(item)

takewhile在满足条件时持续取值：

from itertools import takewhile print(list(takewhile(lambda x: x < 100, fib())))

这三个组合起来，处理数据流时能把“取一部分、过滤一部分”的逻辑写得非常清爽。

4. 进阶使用与问题排查：send、流水线和避坑清单

4.1 send、throw、close：生成器进阶三件套

生成器不只是“产出数据”，还能接收数据。这就是send做的事。看这个累加器：

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

这里yield total不只是一个产出语句，它的返回值是send传进来的值。value = yield total是标准写法，含义是：把当前累加结果yield出去，然后挂起等待接收新值。

使用方式：

acc = accumulator() print(next(acc)) # 0，执行到yield位置 print(acc.send(10)) # 10，yield表达式返回10，total累加后再次yield print(acc.send(20)) # 30

有一个硬性规定：生成器第一次启动时，只能next(acc)或acc.send(None)，直接acc.send(10)会抛TypeError: can't send non-None value to a just-started generator。原因是生成器还没执行到yield表达式，根本没有“接收值的位置”。我测试过多次，这个报错在新手里出现频率极高，建议记住。

throw用于在生成器挂起点注入异常，close用于主动关闭生成器：

g = gen() next(g) g.throw(ValueError, "手动抛入异常") # 在挂起点抛出异常 g.close() # 在挂起点抛GeneratorExit，生成器结束

throw在协程间的错误传递场景有用，close实际用的不多。但理解close的原理很关键：它会在生成器挂起点抛入GeneratorExit，如果你在生成器里写了try/finally，finally块会正常执行，资源可以安全释放。这就是生成器版本的“上下文管理器”。

4.2 常见错误速查表：迭代器与生成器使用避坑指南

第一行的问题最常见。很多人在函数里生成一个生成器，然后传给两个处理函数，第二个函数拿到空数据还一脸懵。记住迭代器是流水线，不是仓库。第二行也高频出现，因为刚接触生成器的人经常想当然地generator[0]取第一个值，正确做法是next(generator)。

4.3 组合成生成器流水线：数据处理的新思路

单个生成器只是惰性计算，把多个生成器串联起来，就变成了管道——数据像水流一样从源头流经各级处理节点，每个节点只负责一件事。

举个日志过滤的例子。串联三个生成器，一个读源数据，一个过滤错误行，一个做结构化解析：

def read_logs(file_path): with open(file_path, "r") as f: for line in f: yield line.strip() def filter_errors(lines): for line in lines: if "ERROR" in line: yield line def parse_error_logs(lines): for line in lines: parts = line.split(" ", 3) yield {"time": parts[0], "level": parts[1], "msg": parts[3]} # 串联 log_stream = read_logs("app.log") error_stream = filter_errors(log_stream) parsed_stream = parse_error_logs(error_stream) for entry in parsed_stream: process(entry)

整个过程，内存里始终只保持一行日志。每一级生成器都可以单独测试，传入一个假数据流看输出是否符合预期，排查问题非常方便。

这种流水线写法比我早期写的“一次处理完再传下一个函数”的方式舒服得多。之前总是一遍遍构造中间列表，数据大一点内存就开始告急，改成一个一个生成器串联后，代码结构反而更清晰，每级职责单一，出了毛病一眼就能定位到是过滤逻辑还是解析逻辑出了问题。

我个人在反反复复折腾过文件解析、数据入库、日志清洗之后，最大的感受是：写Python代码时，只要发现自己循环里不断append一个列表再传给下一个函数，就该停下来想想能不能改成yield。生成器的思维本质上就是“让数据流起来”，一次处理一个元素，不囤货、不积压，内存也就稳了。最后再分享一个小习惯：调试生成器时不要只靠print，多试试list(islice(gen, n))截取前几个值观察输出，比眼睁睁看着生成器无限跑下去高效得多。