大概是两年前我接了一个特别磨人的数据导出需求某个第三方 API 的数据得按页拉每页 50 条处理完这一页再去拉下一页中间还时不时超时断连。第一版我老老实实写了个 while 循环手里维护着页码、游标、重试次数代码越写越长边角情况越补越多最后自己看着都头疼。后来我把整段拉数逻辑封装成一个自定义迭代器调用方那边的十来行循环瞬间缩成一行,那个对象自己知道怎么一页一页往外吐数据、什么时候该停。自定义迭代器这件事,本质上是让遍历从语言内置的 list、dict 扩散到你自己的任何类上什么时候取数、怎么取、取到什么时候算完全由数据结构自己说了算。这篇文章我会从迭代器协议讲起用一个分页 API 的真实案例带你完整实现一遍然后再聊无限序列、惰性求值、常见陷阱和工程上的选型权衡。适合写过一阵 Python、但还没有亲手设计过迭代器的开发者读完你至少能避开我当年踩过的那些坑。1. 自定义迭代器的价值从能遍历到会遍历1.1 迭代器协议解决的三件事很多人对迭代器的理解停留在让我的类能被 for 循环消费这一层但它的价值其实更靠里。我可以负责任地说迭代器协议真正解决了三件事。第一件事是遍历逻辑和数据结构的解耦。一个类内部是数组、链表还是数据库游标外部消费者完全不关心。调用方只需要 for item in obj 就能拿到数据至于数据是从内存里读的还是逐行读文件的都不影响消费端代码。这带来的直接好处是你可以后来把内部实现整个换掉调方一行不用改。第二件事是惰性计算。迭代器不是一次性把结果全部算好塞给你而是你问一次它算一次。对于分页 API、大文件、无限序列这种天然就是持续推进的数据源这一点几乎是刚需。否则你得先把所有数据攒成一个超大 list内存和时间双重爆炸。第三件事是让遍历语法统一化。Python 里 for、列表推导、解包、sum()、max() 这些工具全都可以消费任何可迭代对象。你只要实现了协议你的类就自动获得了整个语言生态里所有吃迭代对象函数的能力。这个杠杆效应比你自己写十几个工具方法划算得多。1.2 一次真实需求从 while 循环到迭代器还是说回我那个分页 API 的场景。传统写法大概是这样的page 1 has_more True results [] while has_more: data api.fetch(pagepage, page_size50) results.extend(data.items) has_more data.has_more page 1 for item in results: process(item)这段代码问题很明显results 把全量数据都装进来了如果总共十万条内存就得装十万条而且拉数据和处理数据两件事被强行绑在一起处理逻辑想提前退出比如查到了目标记录就想停下来也没办法因为你已经把后面几十页全拉完了。改成自定义迭代器之后调用方大概是这个样子for item in PaginatedAPI(https://api.example.com/items): process(item) if item.id target_id: break注意 break 这个操作的语义变化——传统写法里你没有这个选项只能全拉完再筛迭代器写法里循环说停就停API 后面的页根本不会去请求。这种按需推进的能力就是自定义迭代器最让我上瘾的地方。2. 迭代器协议拆解Python 和 JavaScript 的底层约定2.1 Python 双下划线协议__iter__ 与 __next__Python 迭代器协议其实就两个方法加一个异常清晰得近乎粗暴。class Countdown: def __init__(self, start): self.current start def __iter__(self): return self def __next__(self): if self.current 0: raise StopIteration value self.current self.current - 1 return value for n in Countdown(5): print(n) # 输出 5 4 3 2 1__iter__ 负责返回一个迭代器对象__next__ 负责给出下一个值没有值了就得抛 StopIteration。for 循环之所以能识别这个类的结束不是靠判断返回值是 None而是靠捕获 StopIteration 异常。也是因为这一点__next__ 里千万不要把 StopIteration 吞掉否则 for 循环会永远跑下去。这里还要补一个概念区分。可迭代对象iterable指的是实现了 __iter__ 的类它代表可以被遍历迭代器iterator是实现了 __iter__ 和 __next__ 的类它代表遍历进行到哪了。Countdown 既是可以迭代的对象也是它自己的迭代器。但实际场景里这两个角色常常由不同的类承担我后面第 5 节会说这个混淆带来的坑。2.2 JavaScript 的 Symbol.iterator同一种思想的另一套表达如果你平时也写前端会发现 JavaScript 的迭代器协议和 Python 是同一个思想的两副面孔。JS 用的是键名为 Symbol.iterator 的方法next() 返回的不是值而是一个包含 value 和 done 的对象。const countdown { current: 5, [Symbol.iterator]() { return this; }, next() { if (this.current 0) { return { done: true, value: undefined }; } return { done: false, value: this.current-- }; } }; for (const n of countdown) { console.log(n); }两种协议对照起来看核心设计是一致的协议要素PythonJavaScript迭代入口__iter__()[Symbol.iterator]()获取下一项__next__()next()结束信号抛StopIteration返回{ done: true }消费语法for...in循环、推导式for...of、展开运算符判断可迭代isinstance(x, Iterable)检查是否存在Symbol.iterator写惯了 Python 再写 JS 时你只要记住结束通知方式不同这一个差异就够了。其他语言生态里C# 的 IEnumerable、Rust 的 Iterator trait、Java 的 Iterator 接口也全都在做同一件事把怎么产生下一个值抽象成一个协议。2.3 容易被忽略的旧式协议__getitem__ 回退这个绝对是我压箱底的冷知识。Python 的 for 循环在对象没有 __iter__ 时会退回去检查有没有 __getitem__。只要实现了 __getitem__并且让它按 index0,1,2,3... 依次返回数据在越界时抛 IndexError这个对象照样能被 for 循环消费。也就是说__getitem__ 可以单方面让一个类成为可迭代对象。class EvenNumbers: def __getitem__(self, index): if index 10: raise IndexError return index * 2 for x in EvenNumbers(): print(x) # 0 2 4 ... 20这个回退机制是历史遗留但它有个实际价值如果你的类本来就实现了下标访问那你可以不额外写迭代器白嫖一轮 for 循环支持。不过我的建议是这只适合顺手的小场景真正做正经设计时还是老老实实写 __iter__因为基于 __getitem__ 的迭代缺少语义边界可读性和可维护性都差一些。3. 分页数据迭代器实战让 API 自己吐数据3.1 先想接口再想实现写任何自定义迭代器之前我习惯先站在调用方的角度把接口定下来。这个分页迭代器我希望调用方拿到的是一个扁平的数据流感知不到分页的存在就像遍历一个普通列表一样。进一步的期望是想中途停止就能停停了下一次还能重新从第一页开始。这个重新遍历的需求值得单独拎出来说。很多人在第一版设计里会让 __iter__ 返回 self但这样迭代器本身是有状态的遍历完了再 for 一次就什么都拿不到了。更好的做法是让被迭代的容器对象和迭代器对象分离容器对象负责存数据和提供多个独立的迭代器每个迭代器负责记录自己的当前位置。传统写法里我说过要维护 page 变量迭代器设计里就把它放进了迭代器实例的状态里。3.2 完整实现与几个决定成败的细节import requests class PaginatedAPI: 一个可迭代对象每次调用 __iter__ 返回全新的迭代器 def __init__(self, base_url, page_size50): self.base_url base_url self.page_size page_size def __iter__(self): return self._Iterator(self) class _Iterator: def __init__(self, api): self.api api self._page 1 self._has_more True self._buffer [] self._index 0 def __iter__(self): return self def __next__(self): if self._index len(self._buffer): if not self._has_more: raise StopIteration self._fetch_next_page() item self._buffer[self._index] self._index 1 return item def _fetch_next_page(self): resp requests.get( self.api.base_url, params{page: self._page, page_size: self.api.page_size}, timeout10, ) resp.raise_for_status() data resp.json() self._buffer data.get(items, []) self._has_more bool(data.get(has_more)) self._page 1 self._index 0这段代码里有三个细节全是实战里用教训换来的。第一个缓冲区用 index 而不是 pop(0)。如果你一开始用 self._buffer.pop(0) 往外吐数据每吐一条列表就要做一次 O(n) 的搬移一页 50 条还好如果一页上千条性能就会明显下滑。用 index 指针标记当前读到哪里代价只有 O(1)。第二个__iter__ 里返回的是 self._Iterator(self)不是 self。这个设计让同一个 PaginatedAPI 对象可以被多次 for 循环每次循环各自持有独立的遍历状态。如果你图省事让 __iter__ 返回 self这个对象就只能从头到尾完整遍历一次第二次 for 循环直接拿到空结果。第三个_fetch_next_page 里每次用 resp.json() 把整页数据读进内存。理论上可以更进一步做成流式解析但对 JSON API 来说这个粒度已经够用真到单页几 MB 才需要考虑流式。3.3 为什么我不直接返回一个 list在这个案例里最直觉的替代方案是写一个 get_all_items() 方法把所有页拉完返回一个大列表。但如果第一页就有你要找的那条记录或者用户只是想看前三条列表方案会白白发出几百个请求。迭代器方案里用户 for 循环里 break 一下请求立刻停止。再算一笔内存账。假设总数据一万条每条一个字典平均 2KB全部塞进列表是 20MB 内存迭代器方案里任意时刻内存里只有一页 50 条大约 100KB。差了 200 倍。数据量再翻几番这个差距就是能不能跑起来的区别了。4. 超越循环无限序列和惰性求值4.1 用迭代器实现斐波那契数列我第一次意识到迭代器价值是做斐波那契数列。如果用函数直接返回列表你得先定好取多少项因为函数必须返回一个有限的东西。但迭代器根本不受这个限制——它可以永远不结束调用方用 itertools.islice 自己决定取多少。class Fibonacci: def __init__(self): self.a, self.b 0, 1 def __iter__(self): return self def __next__(self): value self.a self.a, self.b self.b, self.a self.b return value from itertools import islice fib Fibonacci() print(list(islice(fib, 10))) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]注意这里限制取多少项的逻辑在迭代器外面通过 islice 完成。这个无限数据源 外部限量的组合模式几乎可以用在任何生成连续数据的场景里时间序列、计数器、随机数流、按序递增的 ID。迭代器把我有哪些数据和你取多少彻底解耦了。4.2 大文件读取惰性收益最直观的场景分页 API 的惰性收益是省内存大文件读取的收益更直观。默认的 for line in open(file.txt) 之所以不会把整个文件读进内存就是因为文件对象本身就是一个迭代器。但很多人不知道 iter() 函数还有一个带哨兵值的用法在处理自定义格式的流式读取时特别好用。def read_until_blank(f): return iter(f.readline, \n) with open(data.log) as f: for block_count, line in enumerate(read_until_blank(f)): # 每读到空行触发一次回调处理一个逻辑块 pass这个 iter(callable, sentinel) 形式每次调用 callable 对象直到返回值等于 sentinel 才停止。它配合 f.readline 可以做到一行一行地喂给下游内存占用恒定在单行大小不管文件是 10MB 还是 10GB这个数字不变。4.3 惰性不是免费的权衡的几个维度惰性求值这么好是不是所有场景都应该用不是。我总结下来要付三个代价。一个是时间上的。迭代器每次求值都有 Python 函数调用开销同样是对一个 10 万元素的列表求和sum(list) 和 for 循环消费迭代器的耗时能差出 10% 到 20%。在数据量小且确定的情况下直接生成列表反而更快。另一个是可读性代价。迭代器的状态是隐式的你没法像看列表那样一眼看出它还有多少数据。调试时想打印还剩几个往往要额外维护计数。还有一个是只能消费一次的硬限制。很多开发者第一次遇到迭代器没数据时会懵其实就是因为迭代器已经跑完了。如果你需要反复遍历同一次数据要么重新生成一个迭代器要么先用 list 固化下来。到底选哪条路取决于你的数据量级和使用次数——数据大、用一次选迭代器数据小、用多次选列表。5. 自定义迭代器踩坑实录四条最常见的事故5.1 迭代器是一次性消费品这是自定义迭代器新手遇到最多的问题特征是第一次 for 循环正常第二次循环什么输出都没有程序也不报错。原因我前面提过——迭代器内部维护着当前位置遍历完了指针就停在结束处不会自动回到开头。it iter(Countdown(3)) print(list(it)) # [3, 2, 1] print(list(it)) # [] # 错误示范直接在同一个对象上 for 两次 cd Countdown(3) print(list(cd)) # [3, 2, 1] print(list(cd)) # []第二次啥也没有解决方案就是我在 3.2 里示范的容器/迭代器分离把数据在哪和读到哪了拆成两个类__iter__ 每次返回一个全新迭代器。这样容器本身可以无限次被遍历每个迭代器各自独立。5.2 边迭代边修改容器RuntimeError 的经典现场这个坑不在自定义迭代器内部而在使用方。如果你在 for 循环里直接往 dict 里加键或者从 list 里删元素Python 会在下一次迭代时抛 RuntimeError: dictionary changed size during iteration。原因是 Python 的容器迭代器按版本号机制检测容器是否被修改过。解决思路有两种。如果需要过滤创建新容器而不是原地删# 错误 # for k in data: # if len(k) 3: # del data[k] # 正确构建新字典 data {k: v for k, v in data.items() if len(k) 3}如果需要真的原地更新就先遍历副本for k in list(data.keys()): if len(k) 3: del data[k]如果你在实现自定义迭代器也要注意你的 __next__ 里如果要修改底层数据最好加一个修改计数每次迭代前比对一下能提前发现使用方的非法操作而不是让程序在奇怪的位置崩溃。5.3 生成器里 StopIteration 的变异这个坑尤其隐蔽。PEP 479 之后生成器内部如果抛出 StopIteration不管是显式 raise 还是无意中让某个迭代器抛出来它不会被当成正常结束信号而是会被转成 RuntimeError。看这个例子def bad_generator(): yield 1 raise StopIteration # 期望结束不RuntimeError!很多人的直觉是 StopIteration 代表结束生成器里手动抛一下也无妨但 Python 3.7 以后这就是个运行时错误。正确的写法是什么都不做让函数自然返回或者用 return 结束def good_generator(): yield 1 return # 这才是显式结束的正确方式这个改动是为了防止生成器内部的迭代器把外层 for 循环的 StopIteration 意外吞掉。你写自定义迭代器时__next__ 里正常抛 StopIteration 没问题但如果你在 __next__ 内部调用了别的迭代器并且放任它的 StopIteration 往外冒这反而是合法且常见的手法不需要改成捕获。5.4 可迭代对象和迭代器边界混乱带来的连锁问题第五个坑源于概念混淆。如果让你的类同时实现 __iter__ 和 __next__且 __iter__ 返回 self那这个类既是 iterable 又是 iterator。这在简单场景没问题比如 Countdown但一遇到嵌套遍历就出错class Bad: def __init__(self, data): self.data data self.i 0 def __iter__(self): return self def __next__(self): if self.i len(self.data): raise StopIteration item self.data[self.i] self.i 1 return item b Bad([[1, 2], [3, 4]]) for row in b: # 第一次循环正常 for x in row: # 但 x 这里拿到的不是数字 pass问题在于内层 for 也会调用 b 的迭代器两个循环共享同一个 self.i 指针数据乱掉只是时间问题。规范的做法是一个类如果同时需要被多次遍历__iter__ 里返回独立的小迭代器对象如果数据本身只能单向消费一次比如流那返回 self 才合理。6. 显式迭代器类还是生成器工程选型的边界6.1 大多数场景下生成器就够用了Python 里的生成器函数就是迭代器的语法糖。凡是能用 yield 写出来的迭代逻辑代码量通常只有显式迭代器类的一半甚至更少def fibonacci(): a, b 0, 1 while True: yield a a, b b, a b我个人的经验是如果我只需要顺序产生一系列值没有复杂的暂停/恢复逻辑无脑用生成器。它天然支持惰性求值写法紧凑调试时还能用 yield 在中间停住观察状态。大概七成到八成的迭代需求都落在这一档。6.2 什么时候必须上显式迭代器类有三类情况我会放弃生成器改回显式类。第一类是状态迁移逻辑比较复杂、需要清晰的属性名来承载状态时。比如实现一个状态机遍历器内部有等待输入、处理中、完成多个阶段生成器的局部变量藏在函数内部没法外部访问调试很痛苦显式类可以把状态写在实例属性上随时打印观察。第二类是同一个数据源需要支持多个独立遍历游标时。我第 3 节的分页 API 就是典型——生成器只有一个游标没法实现有人从第 1 页读同时另一个人从第 5 页读。显式类让每个迭代器实例独立持有 _page 属性天然支持并发遍历。第三类是希望迭代器支持回退、重置这类额外操作时。标准迭代器协议只有前进和结束两个动作但你可以给显式类加方法扩展比如 peek() 看一眼下一个值但不消费或者 reset() 把指针拨回开头。6.3 yield from组合生成器的隐藏利器如果你决定用生成器还有个操作你大概率会用得上——yield from。它可以把子生成器的产出直接透传给外层。最常见的场景是展平嵌套结构def flatten(items): for item in items: if isinstance(item, (list, tuple)): yield from flatten(item) else: yield item print(list(flatten([1, [2, [3, 4]], 5]))) # [1, 2, 3, 4, 5]没有 yield from 的版本你得自己 for 循环子生成器再 yield。有了它委托关系一目了然。这个语法在处理迭代器组合时特别好用相当于给生成器提供了类似函数调用的能力。7. 组合迭代器与 itertools别什么都自己造7.1 标准库的迭代器胶水自定义迭代器最大的价值不是取代 itertools而是和它配合。itertools 提供了一批专门吃迭代器、吐迭代器的函数把它们组合起来能完成非常多看似复杂的工作。from itertools import islice, chain, takewhile, tee fib fibonacci() first_ten list(islice(fib, 10)) # 限量取 even_fib list(islice((x for x in fib if x % 2 0), 10)) # 过滤限量 combined list(chain([0], first_ten)) # 拼接 # tee 把一个迭代器复制成多个独立游标 it1, it2 tee(fibonacci(), 2) # takewhile 在条件不满足时停止 small list(takewhile(lambda x: x 100, fibonacci()))以 tee 为例很多人在又要数人头又要逐条处理的场景里被迭代器只能消费一次卡住tee 直接把它复制成两个独立迭代器一个用来计数一个用来处理。虽然底层做了缓冲内存开销比单次遍历高但比你把全部数据转成列表再遍历还是低得多。7.2 组合迭代器的一个实用案例组合不只是在 itertools 内部做。你自己写的迭代器也可以成为组合链的一部分。比如前面的 PaginatedAPI 和 takewhile 组合可以实现拉数据直到遇到某条件才停from itertools import takewhile api PaginatedAPI(https://api.example.com/items) recent takewhile(lambda item: item.created_at cutoff, api) for item in recent: process(item)更典型的应用是流式处理管道。假设平台每天产生大量事件日志你要做筛选、字段提取、聚合三个步骤每一步都可以包装成迭代器函数最后用 for 循环串联起来。整个处理过程内存占用恒定因为每个环节都是一边读一边吐。7.3 性能实测记录迭代器不是银弹最后交个底说说我自己做过的简单基准测试。对一个 100 万元素的整数序列求和直接 sum(range(1_000_000)) 大概耗时 25ms自己写一个生成器函数再 for 循环累加耗时大约在 70ms 上下如果我用元组把它全部物化再 sum耗时在 45ms 左右。结论很明确迭代器的惰性求值带来的是内存收益不是时间收益每次求值都附带了函数调用开销在纯计算密集场景里反而更慢。所以我的选型原则也很朴实数据源是外部的、不确定量的、可能很大的用迭代器数据源是内存里的小型结构、要反复用的直接列表。别为了用迭代器而用迭代器它解决的是内存和组合问题不是速度问题。最后再分享一个我自己的习惯。写自定义迭代器时我一般先写一个最小的生成器版本把逻辑跑通确认数据流是对的再根据是否需要多游标、需要重置、需要外部观察状态来决定要不要重构成显式类。顺序反过来很容易过度设计明明一个 yield 就搞定的事硬是写了两个类。另外强烈建议在迭代器里把结束条件写在显眼的位置并且加清楚的注释——我见过太多迭代器 bug都是结束条件藏在某个深层的 if 里别人包括几个月后的自己压根看不出来它到底什么时候停。迭代器这东西设计时多花十分钟想清楚协议边界使用时能省下后面无数个排查的夜晚。