Python自定义迭代器实战:从分页API到惰性求值
2026/9/9 4:47:42 网站建设 项目流程

大概是两年前,我接了一个特别磨人的数据导出需求:某个第三方 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(page=page, page_size=50) 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__,并且让它按 index=0,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_size=50): 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}, timeout=10, ) 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 里,别人(包括几个月后的自己)压根看不出来它到底什么时候停。迭代器这东西,设计时多花十分钟想清楚协议边界,使用时能省下后面无数个排查的夜晚。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询