Python迭代器与生成器：彻底搞懂for循环底层机制与实战

咱们写 Python 写了这么久几乎每天都在用for循环遍历列表、字典、文件、range但有没有一瞬间好奇过for循环凭什么能遍历这么多长得完全不一样的东西它内部到底是怎么知道“下一个元素”是什么的又是怎么判断“已经到最后了”的这个问题往深挖就是迭代器Iterator。说白了Python 里的for循环只是一个语法糖真正干活的是迭代器协议。今天这篇就围绕“迭代器”把for循环背后的故事讲透——我会从最基础的概念讲起手写一个迭代器再聊到生成器、itertools进阶玩法最后还会分享一个实战级的大批量数据加载器写法。整个过程我会把踩过的坑和查问题时的小技巧一并写出来适合刚学 Python 不久、想搞懂底层机制的人也适合写了不少代码但一直被“迭代器一次性消费”这类问题折磨的老手。1. 迭代器是什么先搞清楚三个概念1.1 可迭代对象、迭代器、生成器别再混淆很多教程会把“可迭代对象”“迭代器”“生成器”混着说但它们在 Python 里其实是三个不同的东西。我刚学的时候被绕晕过后来用一句话记可迭代对象是“能提供迭代器的东西”迭代器是“负责一个一个往外掏东西的东西”生成器是“用yield写出来的迭代器”。Python 官方的定义更严格可迭代对象iterable实现了__iter__()方法调用它返回一个迭代器。比如列表、字典、集合、字符串、文件对象。迭代器iterator实现了__iter__()和__next__()两个方法的对象。__next__()每次返回下一个元素没有元素时抛StopIteration异常。生成器generator是迭代器的一种由包含yield的函数调用后产生内部自动实现了__iter__()和__next__()。用生活里的例子比划一下列表是一本纸质书书本身是一本完整的书可迭代对象但它不能自己翻页iter(列表)相当于给你一个书签迭代器书签记录当前读到哪每次next()就是翻一页而生成器像是一个自动翻页的电子书签它自己记得读到哪还能临时掐断、下次继续。写代码验证就更直观了lst [1, 2, 3] print(hasattr(lst, __iter__)) # True列表是可迭代对象 print(hasattr(lst, __next__)) # False列表不是迭代器 it iter(lst) print(hasattr(it, __iter__)) # True迭代器也是可迭代对象 print(hasattr(it, __next__)) # True这里有个容易忽略的细节迭代器本身也是可迭代对象它的__iter__()通常返回自己。这意味着你可以在迭代器上继续调用iter()得到的还是同一个迭代器。1.2 for循环到底做了什么for item in iterable这行代码拆开看等价于下面的while循环it iter(iterable) while True: try: item next(it) except StopIteration: break # 循环体 print(item)也就是说for循环做了三件事调用iter(iterable)拿到一个迭代器对象不断调用next(iterator)取出下一个元素捕获StopIteration异常后正常结束循环。所以任何对象只要实现了__iter__()或者__getitem__()理论上都可以被for循环遍历。Python 的老式双协议设计到现在还保留着如果一个类没有__iter__()Python 解释器会退回去走“索引协议”也就是从下标 0 开始不断调用obj[0]、obj[1]直到抛出IndexError才结束。这就是为什么某些老式对象也能被for遍历。理解了这一点你就明白为什么字符串能遍历、为什么int不能遍历int既没有__iter__()也不能按下标访问所以for i in 123会直接抛TypeError。我建议你以后写自定义类时别偷懒只实现__getitem__尽量显式实现__iter__和__next__因为很多时候代码要兼容、要判断isinstance(obj, Iterable)这种检查在只有__getitem__的对象上会得到False容易埋坑。2. 手写一个迭代器迭代器协议到底怎么实现2.1iter和next的配合规则既然for依赖迭代器协议那我们自己造一个迭代器就顺理成章了。规则不复杂__iter__(self)返回一个迭代器对象。如果类本身是迭代器通常return self__next__(self)返回下一个元素。如果没元素了必须抛StopIteration。来看一个最低限度的例子模拟从start数到endclass RangeIterator: 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 for i in RangeIterator(1, 5): print(i, end ) # 输出1 2 3 4每次调用next()current自增状态保存在实例属性里。当current越界时抛StopIterationfor循环收到后正常退出。这里有一个写类迭代器时非常常见的问题把__iter__写成return self会导致“同一个迭代器只能完整遍历一次”。如果你想要一个可迭代对象支持多次遍历应该让__iter__()每次返回一个新的迭代器对象。常见的做法是拆成两个类容器类只实现__iter__迭代器类单独实现__next__。举个例子class MultiRange: def __init__(self, start, end): self.start start self.end end def __iter__(self): return RangeIterator(self.start, self.end) for i in MultiRange(1, 4): print(i) for i in MultiRange(1, 4): print(i) # 正常第二次遍历如果直接在RangeIterator里返回 self第二次遍历就什么都拿不到了。这个“迭代器是一次性物品”的特性后面还会再讲。2.2 完整示例斐波那契迭代器写一个斐波那契迭代器是练习迭代器协议最经典的题目因为斐波那契数列天然是“按需生成、永不回头”的流。class FibonacciIterator: def __init__(self, max_countNone): self.a, self.b 0, 1 self.max_count max_count self.count 0 def __iter__(self): return self def __next__(self): if self.max_count is not None and self.count self.max_count: raise StopIteration value self.a self.a, self.b self.b, self.a self.b self.count 1 return value fib FibonacciIterator(10) for num in fib: print(num, end ) # 0 1 1 2 3 5 8 13 21 34如果不指定max_count那这个迭代器是无穷的直接for会把机器跑死。所以用迭代器的时候一定要想清楚终止条件。这也是为什么很多迭代器设计都要带“上限”参数不是没用的。如果你只想取前几个可以配合itertools.islicefrom itertools import islice fib FibonacciIterator() first_20 list(islice(fib, 20))2.3 为什么用迭代器模式而不是直接给列表加索引有人会问for直接用下标i 0; while i len(lst): lst[i]不就行了为什么还要搞出一套__iter__和__next__核心原因有三个内存、延迟、统一接口。以range(10**9)为例如果把它变成列表内存瞬间爆炸但迭代器只需要存一个current值随取随丢。这就是“惰性计算”的思想。再比如文件对象本质上就是一个“生成行的迭代器”如果先readlines()读成全量列表一个大几 GB 的日志文件能把电脑卡死如果逐行迭代内存占用几乎是零。统一接口也极其重要。Python 里的map、filter、zip、enumerate都返回迭代器它们可以被随意组合、流水线处理而底层的实现五花八门全靠同一个协议往上对接。这就是面向协议编程的威力和接口多态是一样的道理。3. 生成器偷懒又强大的迭代器3.1 yield 让普通函数变成生成器手写一个类实现__next__虽然不难但代码还是啰嗦。Python 给了我们一个更偷懒的武器yield。只要一个函数内部出现了yield调用这个函数就不会执行函数体而是返回一个生成器对象。生成器自动实现了迭代器协议。def countdown(n): while n 0: yield n n - 1 gen countdown(3) print(gen) # generator object countdown at 0x... print(next(gen)) # 3 print(next(gen)) # 2 print(next(gen)) # 1 # 再 next(gen) 会抛 StopIterationyield的作用是把当前函数的状态冻结住包括局部变量、指令指针、甚至执行到一半的表达式等下次调用next()的时候再恢复。你可以把它想象成一个可以随时暂停和恢复的电影yield就是暂停键。这里要强调一个新手特别容易踩的坑生成器函数一旦包含yield函数体不会立刻执行。所以如果你想在函数一开始做检查、打日志指望调用函数时就触发那是不行的。得等第一个next()才会真正开始执行。调试的时候这一点特别容易让人懵。与传统迭代器类相比生成器省去了__iter__、__next__和StopIteration的手工管理代码短得惊人非常适合表达“流”概念。比如逐行读文件def read_lines(path): with open(path, r, encodingutf-8) as f: for line in f: yield line这段代码已经是很标准的写法了大文件也能扛得住。3.2 生成器表达式列表推导式的低配版生成器表达式和列表推导式长得几乎一样只是把中括号换成小括号squares_list [x * x for x in range(10)] # 列表立刻算出全部 squares_gen (x * x for x in range(10)) # 生成器惰性求值 print(type(squares_list)) # class list print(type(squares_gen)) # class generator如果只是把它传给sum()、max()这种“消费一次”的函数生成器表达式非常省内存。比如算前 1000 万个数的平方和total sum(x * x for x in range(10_000_000))换成列表推导式会先创建一个巨大的列表再求和内存峰值会高很多。生成器表达式一路流式算完基本不占内存。但要注意生成器不能重复使用。你要把同一个生成器传给两个函数第二个函数会拿到空数据。所以如果一组数据要多次遍历、或者要随机访问老老实实转成列表或用列表推导式别硬用生成器。3.3 yield from 与生成器委托yield from是 Python 3.3 引入的语法用来把子生成器的产出“透传”给外层生成器。一个典型例子是自己实现chain函数def chain(*iterables): for it in iterables: yield from it for item in chain([1, 2], ab): print(item) # 输出 1 2 a b如果你写的是嵌套的for ... in ...: yield xxx很多时候可以直接换成yield from代码会清爽不少。它还能把返回值从子生成器传回来不过日常用到不多。生成器还有send(value)、throw(exc)、close()方法可以双向通信。send是协程的基础当你看到yield出现在赋值右边时说明写的不是简单迭代器而是协程了。这篇文章先点到为止等以后写协程专题再展开。4. 内置迭代器工具与进阶玩法4.1 itertools 里最常用的几个函数itertools是 Python 迭代器的“军火库”里面每个函数返回的都是迭代器。我最常用的几个列在下面函数作用示例itertools.count(start, step)无限等差数列islice(count(2, 3), 4)得到2 5 8 11itertools.cycle(iterable)无限循环遍历islice(cycle(AB), 5)得到A B A B Aitertools.repeat(obj, times)重复对象repeat(x, 3)得到x x xitertools.chain(*iterables)串联多个迭代器chain([1,2], [3])得到1 2 3itertools.islice(it, stop)对迭代器切片/截取islice(range(100), 10)取前 10 个itertools.takewhile(pred, it)只要条件满足就产出takewhile(lambda x: x 3, [1,2,3,1])得到1 2itertools.compress(data, selectors)按布尔序列筛选compress(ABCD, [1,0,1,0])得到A Citertools.groupby(it, keyNone)连续分组用于按 key 分组相邻元素几个使用场景一起说。配合无限迭代器取前几个值from itertools import islice, count for i in islice(count(10, 5), 3): print(i) # 10 15 20合并多个数据源from itertools import chain for item in chain([1, 2], {x: 1, y: 2}, ab): print(item) # 1 2 x y a bgroupby有个典型注意事项它只能对已经按相同 key 排好序的连续元素分组不是全局分组。如果数据没有排好序要先sorted()再分组。4.2 反向迭代、多迭代器并行与无限迭代器reversed(obj)可以反向迭代但它要求对象有__reversed__()或支持序列协议有长度和下标。reversed(range(5))没问题reversed(iter([1,2,3]))不行因为普通迭代器没有长度。zip(it1, it2, ...)返回一个迭代器每次从每个传入的迭代器里各取一个得到元组。zip在其中一个迭代器耗尽时就会停不需要手动判断长度。enumerate(iterable, start0)本质上是给迭代器“编号”。Python 3 的map()和filter()也返回迭代器不是列表这一点和 Python 2 不一样写代码时别默认能直接len()。无限迭代器是迭代器模式的野蛮体现。比如itertools.count()无限递增如果不加终止条件list(count())会直接内存溢出。安全写法是配islice或手写break。from itertools import count, islice for i in islice(count(), 5): print(i) # 0 1 2 3 44.3 自建迭代器与 iter() 的第二参数用法iter()除了常规用法还有一个隐藏技能iter(callable, sentinel)。第一个参数是无参函数第二个参数是哨兵值。每次迭代会调用函数取返回值当返回值等于sentinel时马上停止。这个在按块读文件时特别好用。例如每次读 16 字节读到空字节就停with open(data.bin, rb) as f: for chunk in iter(lambda: f.read(16), b): process(chunk)也可以用来从输入里读行直到遇到结束标记def read_until_end(): return input(输入 exit 退出: ) for line in iter(read_until_end, exit): print(f处理: {line})新手看到这种写法会觉得很怪但它确实比 while 循环更像 Python。等你会用iter(callable, sentinel)之后很多“排队轮询”“分批拉取”的循环代码都能写得非常干净。5. for循环背后的常见坑与调试技巧5.1 迭代器是一次性消耗品列表是可迭代对象重复for很多次都没问题但迭代器是“单程票”。比如it iter([1, 2, 3]) print(list(it)) # [1, 2, 3] print(list(it)) # []因为迭代器已经耗尽这个坑最常见的场景是把迭代器存成属性或变量第一次用完后忘了重新生成第二次循环发现是空的还以为是数据丢了。解决办法是如果你要让一个对象支持多次遍历不要在__iter__里return self而是每次返回一个新的迭代器如果你已经有一个迭代器并且需要多次使用请提前list()保存或者重新调用能生成新迭代器的函数。5.2 迭代时修改集合会翻车在for循环里直接修改正在遍历的列表是很多新手必踩的雷。看这段lst [1, 2, 3, 4, 5] for item in lst: if item % 2 0: lst.remove(item) print(lst) # 预期 [1,3,5]实际 [1, 3, 5]? 不一定看运气因为迭代器内部用下标或游标记录位置你删除元素后后面的元素会往前移循环可能会跳过某个元素。更糟糕的是遍历字典时增删元素会直接抛RuntimeError: dictionary changed size during iteration。正确做法遍历副本for item in lst[:]或for item in list(lst)先收集要删除的条件循环结束后统一删除如果只是生成新列表用列表推导式或filter。遍历副本虽然多占一点内存但安全第一。对超大型列表如果内存吃紧可以倒着遍历删除或者用标志位延迟删除。5.3 StopIteration与next默认值手动调用next(it)时迭代器一旦耗尽就会抛StopIteration。很多时候我们不想处理异常可以直接传默认值it iter([1, 2]) print(next(it)) # 1 print(next(it, None)) # 2 print(next(it, 0)) # 0没有元素时返回默认值注意用next(it, default)时如果迭代器确实没有下一个元素它会返回默认值而不会抛异常。这在消费“可变长数据流”时特别有用。还有一个和生成器相关的坑来自 PEP 479在生成器内部如果StopIteration未被捕获直接冒泡会被转换成RuntimeError。也就是说你不能在生成器里用“让迭代器自然耗尽”来作为结束信号如果有这种逻辑要显式return。这是 Python 3.7 之后的规则老项目升级后容易遇到。5.4 调试迭代器用 tee、list、length_hint检查状态迭代器没法直接看“内容”所以调试时我一般用这几种手段。最简单list(it)把剩余部分全部变成列表直接打印。代价是可能会改变迭代器所处状态它会消费掉迭代器。想看“当前剩余量”可以用it.__length_hint__()返回一个估计值不保证准确但能帮你判断迭代器是不是快空了。想并行处理多个消费者可以用itertools.tee(it, n)把一个迭代器“复制”成 n 个独立迭代器。注意tee会缓存数据内存上不是免费的别滥用。想确认生成器执行到哪一行可以在关键位置加print(fyield {value})或者干脆用next(gen)一步步调试。我调试生成器时也常用debug参数开关日志def gen_data(data, debugFalse): for item in data: if debug: print(fyield {item}) yield item写代码时加一两行日志排查问题能省很多时间。6. 实战案例写一个批量数据加载迭代器6.1 背景大文件/数据库分批读取真实业务里经常要处理“大得不能一次性读进内存”的数据几百万行的 CSV、几千万条数据库记录、训练集图片路径等。如果统统read()或list(db.query(...))轻则内存紧张重则直接卡死。通用的优化思路是分批batch处理每次从数据源取一小批处理完一批再取下一批。这个模式很适合用迭代器封装。我们要实现一个BatchIterator它包装任意可迭代对象按照指定batch_size批量产出列表。6.2 实现BatchIterator直接上代码class BatchIterator: def __init__(self, iterable, batch_size): self._it iter(iterable) self.batch_size batch_size def __iter__(self): return self def __next__(self): batch [] for _ in range(self.batch_size): try: batch.append(next(self._it)) except StopIteration: break if not batch: raise StopIteration return batch用法很简单for batch in BatchIterator(range(10), 3): print(batch)输出[0, 1, 2] [3, 4, 5] [6, 7, 8] [9]最后的批次只有 1 个元素这是允许的。如果你希望最后一笔不足时也丢弃可以在__next__里对len(batch) batch_size做校验。我通常选择保留尾批因为数据处理时不想丢数据。这段代码里有几个设计细节值得说构造时传入iterable但不要在构造时立刻iter()也可以推迟到第一次__next__。我给的是立即iter()好处是可以尽早发现传入对象不可迭代的问题。try ... except StopIteration是必须的因为底层迭代器很可能在批次中间耗尽。__iter__返回 self所以这个批次迭代器只能用一次。有些框架中数据需要 repeat那应该在批量外层再加一层“每次重新创建迭代器”的逻辑。更偷懒的写法是利用iter(callable, sentinel)和islicefrom itertools import islice def batch_loader(iterable, batch_size): it iter(iterable) return iter(lambda: list(islice(it, batch_size)), [])这个写法很简洁但异常处理和可读性不如类。两种都记住按团队风格选。6.3 结合文件读取与缓存优化现在把BatchIterator用在大文件读取上。假设有一个 CSV 文件每行是一个样本你想批量读取import csv def read_csv_rows(path): with open(path, r, encodingutf-8) as f: reader csv.reader(f) yield from reader for batch in BatchIterator(read_csv_rows(data.csv), 1000): # 每批最多1000行 process(batch)这里csv.reader本身返回迭代器yield from把它透传出去。内存里永远只保存当前 1000 行不会把整个文件装进内存。如果数据来自数据库很多数据库驱动的游标本身支持迭代可以直接把游标包进BatchIterator。例如cursor connection.cursor() cursor.execute(SELECT ...) for batch in BatchIterator(cursor, 500): save_batch(batch)如果cursor不可重复迭代每次组件重新查询后再包一层生成器即可。还可以在批量迭代器里加入缓存、预处理、异常重试等逻辑。比如对某条数据解析失败可以记录日志继续不中断整个流。因为BatchIterator只是把数据处理管道的一部分它不关心上游数据从哪来这样代码的复用性会非常好。最后分享一点我的经验我自己在实际项目里踩过不少迭代器的坑最有价值的一条经验是把数据处理管道想成一串迭代器而不是一节一节的列表。数据从文件、数据库、网络来经过map、filter、batch等变换最后交给消费端中间全程惰性求值。这样写出来的程序内存可控、结构清晰也更容易做单元测试。另一个小技巧是如果你在写一个自定义容器类先别急着写__len__、__getitem__先想想迭代器协议能不能解决问题。很多时候一个生成器函数比一整类简化得多。希望这篇把for循环背后的故事讲清楚后你再看 Python 里的循环和数据流会有完全不一样的感觉。