1. 为什么自定义迭代器值得单独拿出来讲很多写了一段时间 Python 的人都有这种感觉列表、字典、集合这些内置类型用得很熟for循环也写得飞起但一旦提到“迭代器”三个字脑子里只剩一个模糊的印象——好像是个能next()的对象再问深一层for循环到底是怎么从列表里把元素一个个取出来的为什么有的对象可以for有的不行为什么有时候同一个对象能遍历两遍有时候却只能遍历一遍这些问题的答案都指向迭代器协议。我自己最初写业务代码的时候对迭代器的理解就是“能用for遍历的东西”完全没想过要自己去实现一个。直到有一次要处理一个很大的日志文件文件里每行是一个 JSON 对象总共几百万行。直接用readlines()读内存肯定爆一行行readline()写起来又啰嗦而且过滤器逻辑和文件读取逻辑混在一起代码很快就变得没法看了。当时我试着写了一个自定义迭代器类把“从文件里读一行 JSON”封装成__next__瞬间整个数据处理的代码变得清爽了而且内存占用稳定在一个很低的值。从那一刻起我才意识到自定义迭代器不是面试题里才用到的东西它是真实工程里非常有用的抽象工具。这篇博文不是讲 Python 内置函数iter()的简单用法而是聚焦在“如何从零设计一个自定义迭代器”。我会先拆解迭代器协议的本质然后从最朴素的类实现讲起再到生成器这种更高级的写法还会聊到双向迭代、无限序列、缓存机制、工程流水线里迭代器的实战价值最后把我踩过的一些坑和设计取舍一并交代清楚。适合的读者是那些已经会用 Python 写基本代码、但想进一步理解语言机制的人以及在实际项目中遇到过“遍历复杂数据源”“内存占用过高”“数据懒加载”等问题的开发者。2. 迭代器协议拆解__iter__和__next__到底在做什么2.1 一个对象能被for循环遍历本质是什么for循环是 Python 里最常见的遍历语法但它背后做事情的方式远比表面看起来更机械。当你写for item in container: print(item)Python 解释器实际上执行了这样几个步骤调用iter(container)尝试获取一个迭代器对象在一个隐藏的循环里反复调用next(iterator)每次拿到一个值就进入循环体直到next()抛出StopIteration异常循环便安静地结束。关键点在于for循环本身并不直接接触列表内部的数据。它接触的永远是“迭代器”。列表、元组、字典这类内置容器Python 会为它们各自生成一个独立的迭代器对象这个对象知道当前遍历到了哪个位置。列表和它的迭代器是两码事是很多新手容易忽略的一点。列表本身是容器它内部存了全部元素而迭代器是一个有状态的对象它只记录“下一个该取哪个元素”。举个例子lst [1, 2, 3] it iter(lst) print(next(it)) # 1 print(next(it)) # 2lst本身没有被消耗lst里的三个元素原封不动在那里。被消耗的是迭代器it它内部维护着一个游标取一次挪一格。取完第三个之后再调用next(it)就会抛StopIteration。2.2 协议对类的要求__iter__和__next__为了让一个自定义的类能被for循环消费它必须满足“迭代器协议”。协议在 Python 里不是强制的接口继承而是一种约定你只需要实现指定的方法解释器就会自动识别。核心方法有两个__iter__(self)返回一个迭代器对象。一个对象如果实现了这个方法它就是“可迭代的”iterable。注意这个方法返回的对象不一定就是self它可以是任何一个实现了__next__的对象。__next__(self)返回下一个元素。如果没有更多元素了必须抛出StopIteration异常。一个类如果同时实现了这两个方法它既是可迭代对象也是自身的迭代器。这种设计在简单场景下很好用但有一个隐蔽的坑同一个对象只能被完整遍历一次。因为迭代器的游标是对象自带的第一轮遍历结束后游标已经走到末尾第二轮直接就是空的。2.3 一个反直觉的机制序列协议的回退Python 判断一个对象是否可迭代并不是严格地看它有没有__iter__。如果对象实现了序列协议中的__getitem__也就是可以通过索引访问元素即使它没有__iter__iter()函数也会自动为它生成一个“索引型迭代器”。这是 Python 为了兼容老代码留下的一个后门机制。class SimpleRange: def __init__(self, start, end): self.start start self.end end def __getitem__(self, index): pos self.start index if pos self.end: raise IndexError return pos这个类没有__iter__也没有__next__但你可以直接for n in SimpleRange(1, 5)遍历它因为解释器发现它有__getitem__就从索引 0 开始依次调用遇到IndexError就认为遍历结束。这个机制本身很有用很多简单的数据结构只需要实现__getitem__就能获得遍历能力但它在排查问题时容易误导人。你往一个类里加了一个魔术方法结果发现它突然变得可以for了但明明没有实现迭代器协议——你得意识到是这个回退机制在起作用。3. 从零实现写一个真正能跑的自定义迭代器3.1 场景设定我要遍历的是什么自定义迭代器最常见的使用场景是遍历那些“不是天然躺在内存里的数据”或者“有多种遍历逻辑的数据”。光讲协议太抽象我们直接做一个有实际意义的例子模拟一个分页 API 的客户端迭代器。假设某外部数据接口采用分页方式返回数据请求 URL 形态为https://api.example.com/v1/pages?page1size100返回的 JSON 结构是{ data: [ {id: 1, name: A}, {id: 2, name: B} ], page: 1, has_more: true }如果你要在业务里把每一页的数据逐个拿出来处理最朴素的做法是写一个循环手动管理页码page 1 while True: resp requests.get(fhttps://api.example.com/v1/pages?page{page}size100) payload resp.json() for item in payload[data]: process(item) if not payload[has_more]: break page 1这段代码能跑但它把“请求网络”和“逐条处理”紧紧地耦合在了一个函数里。如果你有多个地方都要遍历这个接口的全部数据就得把这段逻辑复制粘贴若干次。而且page变量、while循环、break的判断全是过程式思维。换一个思路把它封装成一个迭代器遍历逻辑就变成了一件“关于数据源的事”处理逻辑只需要写一遍。3.2 朴素版实现状态机放在__next__里第一版不需要任何高级技巧直接在一个类里维护三个状态变量当前页码当前页内索引当前页的数据列表、是否还有更多页import requests class PagedAPI: def __init__(self, base_url, page_size100): self.base_url base_url self.page_size page_size self.current_page 0 self.items_in_page [] self.item_idx 0 self.has_more True def _fetch_page(self): resp requests.get( self.base_url, params{page: self.current_page, size: self.page_size} ) resp.raise_for_status() payload resp.json() self.items_in_page payload[data] self.has_more payload[has_more] def __iter__(self): return self def __next__(self): if self.item_idx len(self.items_in_page): if not self.has_more: raise StopIteration self.current_page 1 self._fetch_page() self.item_idx 0 if not self.items_in_page: raise StopIteration item self.items_in_page[self.item_idx] self.item_idx 1 return item使用方式变得非常简洁for item in PagedAPI(https://api.example.com/v1/pages, page_size100): process(item)写这个方法的时候有几点值得注意。__next__里必须先判断“当前页的索引是否已经越界”如果越界再判断“是否还有下一页”。如果当前页刚好取完最后一个元素item_idx等于列表长度这时候进入下一轮循环先看还有没有下一页有就拉取下一页并重置索引没有就直接抛StopIteration。这里有一个边界情况最后一页如果data是空的按照我的实现会触发if not self.items_in_page的兜底逻辑直接抛StopIteration结束遍历。这个迭代器把网络请求延迟到了真正需要数据的时候才发生这就是“懒加载”的核心思想。创建PagedAPI对象的时候不会发出任何 HTTP 请求只有当你开始for遍历元素不够用了它才去下一页拉数据。3.3 为什么__iter__返回self是有道理的上面代码里__iter__直接返回self意味着你把 PagedAPI 对象本身当成了自己的迭代器。这符合“迭代器是可迭代的”这一惯例但有个副作用同一个实例一旦被完整遍历过一遍就废了。api PagedAPI(https://api.example.com/v1/pages) list1 list(api) # 拉完了所有页 list2 list(api) # 直接就是 []如果你需要支持“可以多次遍历同一个数据源”更合适的设计是让__iter__每次返回一个新的迭代器对象而不是返回self。例如class PagedAPIContainer: def __init__(self, base_url): self.base_url base_url def __iter__(self): return PagedAPI(self.base_url)容器只有__iter__每次被for的时候都现场创建一个全新的PagedAPI迭代器遍历完就丢掉下次再遍历重新来一遍。这就是“可迭代对象”和“迭代器”各自的角色分工容器负责描述数据源迭代器负责记录遍历状态。这个设计取舍很重要。很多业务场景里“数据只能遍历一次”是可以接受的甚至就是期望的行为——比如流式处理一个大文件你并不想反复重读。但如果你要实现的是一个可以反复遍历的内存集合那就必须可控迭代器和容器分离。4. 生成器用yield把迭代器写得像同步代码4.1 生成器函数到底做了什么手动写__iter__和__next__的方法在简单场景下还行一旦状态转移比较复杂代码就变得有些冗长。比如上面那个分页 API 的迭代器真正核心的逻辑其实就两三行拉数据、逐条 yield。Python 提供了一种更符合直觉的写法——生成器函数。一个函数里只要出现yield关键字它就不再是普通函数而是生成器函数。调用它不会立刻执行函数体而是返回一个生成器对象。生成器对象是迭代器协议的内置实现你不需要手动维护游标状态解释器会在yield的位置帮你“挂起”和“恢复”。用生成器重写分页 APIdef iter_paged_api(base_url, page_size100): page 0 while True: resp requests.get( base_url, params{page: page, size: page_size} ) resp.raise_for_status() payload resp.json() for item in payload[data]: yield item if not payload[has_more]: break page 1这个函数读起来就像同步代码一样自然你写一个while循环拉数据yield一条拉完就break。生成器自动保存局部变量page自动暂停和恢复执行流你完全不需要关心item_idx之类的状态变量。对比一下生成器版本和类版本的差异非常明显对比维度手动类迭代器生成器函数状态管理需要自己维护变量解释器自动保存代码结构状态判断分散在多个方法里线性流程像普通函数可读性中等逻辑多时容易晕高数据流一目了然支持 send 传值需额外设计原生支持实现难度较高极低4.2 生成器不等于 itertools别混淆这里要澄清一个概念。很多教程会把itertools和生成器放在一起讲但它们是不同层面的东西。itertools是标准库里的一个工具模块里面提供的chain、groupby、islice、product这些函数绝大部分返回的是迭代器对象很多内部就是用 C 语言写的高效生成器。而生成器是你自己用yield写出来的自定义迭代逻辑。两者可以协同工作你可以把自己的生成器传给itertools的函数组合出更复杂的流水线。举个例子假如你想两页一取交错合并两个分页接口的数据import itertools api_a iter_paged_api(https://api.example.com/v1/a) api_b iter_paged_api(https://api.example.com/v1/b) for item_from_a, item_from_b in itertools.zip_longest(api_a, api_b): if item_from_a is not None: process(item_from_a) if item_from_b is not None: process(item_from_b)zip_longest会同时驱动两个迭代器直到两个都耗尽为止。两个生成器各自维护自己的分页状态互不干扰数据量大的场景下内存占用始终是常数级别的这种组合能力是手写类迭代器很难达到的。4.3 什么时候仍然选择手写类而不是生成器生成器虽然好用但并不是万能的。有几个场景我建议你回头用手写类第一当迭代器需要和外部系统交互、需要明确的close()清理资源时。生成器对象虽然有close()方法能在yield点抛出GeneratorExit但无论如何不如类里写一个__del__或显式的cleanup()来得清晰。第二当遍历逻辑需要频繁在“迭代”和其他操作之间切换时。生成器是一种“半协程”只能从上次yield的地方继续没法在迭代过程中从外部向内部发起额外的调用。如果你需要在遍历过程中提供“跳过”“回退”“暂停继续”等复杂控制手写类配合方法调用会更直观。第三性能极端敏感的场景。虽然是 CPython 里生成器的开销已经很小但每一次yield都是一次函数调用级别的成本在每秒几百万次迭代的核心循环里手写类配合局部变量优化可能更可控。不过说实话绝大多数业务场景都到不了这个量级优先考虑可读性就好。5. 进阶设计双向迭代、无限序列、缓存与失效5.1 双向迭代别硬套迭代器协议迭代器协议本质上只要求__next__返回下一个元素它天然是单向、一次性的。如果你需要从一个数据集合里既能往前又能往后有两条路可以走。第一条路无视迭代器协议直接设计一个独立的类提供next()和prev()方法让外部代码显式调用。这种设计最直观也最容易理解缺点是没法直接用for循环糖。第二条路在迭代器协议上叠加一个历史缓冲区记录已经遍历过的元素然后提供一个prev()方法让用户可以回到之前的位置。这个缓冲区可以是无限增长的列表也可以是有固定上限的滑动窗口。我给一个带回溯缓冲的迭代器设计思路class BacktrackableIterator: def __init__(self, iterable): self._src iter(iterable) self._buffer [] # 已经取出过的元素 self._cursor 0 # 当前游标指向下一个要返回的元素位置 def __iter__(self): return self def __next__(self): if self._cursor len(self._buffer): item self._buffer[self._cursor] self._cursor 1 return item item next(self._src) self._buffer.append(item) self._cursor 1 return item def prev(self): if self._cursor 0: raise IndexError(cannot go before the first item) self._cursor - 1 return self._buffer[self._cursor] def reset(self): self._cursor 0这个类的核心思路是底层源迭代器负责真正产生新元素已经产生过的元素全部暂存在_buffer里prev()只是把游标往回拨一下。如果往前拨到的位置已经出过元素就直接从缓冲区拿不会再向底层源要新数据。这种“基于迭代器 缓冲”的回溯方案在遍历流式数据时需要小心如果数据源是网络流或文件流底层迭代器推进之后就无法回头缓冲是实现回退的唯一手段所以缓冲区可能越来越大。解决方式是给缓冲区设置上限当回退深度超过上限时直接报错或提供截断策略。5.2 无限序列迭代器最惊艳的用法之一迭代器是唯一能在 Python 里优雅表示“无限序列”的方式因为它的计算是惰性的永远只计算下一个要的元素而不需要一次性把所有元素加载进内存。我来写一个素数生成器这是无限序列的经典案例def primes(): yield 2 seen [] candidate 3 while True: is_prime True limit int(candidate ** 0.5) 1 for p in seen: if p limit: break if candidate % p 0: is_prime False break if is_prime: seen.append(candidate) yield candidate candidate 2这个生成器可以一直迭代到任意大的素数但你不用担心内存爆炸——seen列表保存的是已经找到的素数用于后续整除判断空间复杂度是 O(已产生的素数个数)这是由“判断素数”这个需求本身决定的不是迭代器的锅。取前 10 个素数只需要import itertools first_ten list(itertools.islice(primes(), 10))islice从无限序列里截取前 10 个然后list把它耗尽。底层生成器的状态被驱动了 10 次之后islice会关闭它primes()内部循环自然就停了。无限序列的使用门槛在于你必须随时有“取够就停”的觉悟。写for p in primes():不带任何终止条件就是死循环。所以实践中无限迭代器场景常和itertools.islice、takewhile配合使用前者按数量截取后者按条件截取。5.3 缓存与失效什么时候迭代器让你“数据过期”自定义迭代器的优势是懒加载但懒加载有一个代价缓存的值可能过期。这里我想说一个真实踩过的坑能帮你避免类似的错误。某次我需要设计一个“按配置加载系统参数”的迭代器。配置本身是从数据库查的且业务运行期间可能有其他进程修改这份配置。我最初写了一个生成器在第一次迭代时从数据库把所有配置项查出来然后yield给调用方。这个生成器本身没有问题但问题出在调用方的使用姿势上——他们在模块加载时就调用了list(config_dict.items())把结果存到了一个全局变量里之后整个进程生命周期内用的都是这份缓存。结果是数据库里的配置改了但运行中的服务始终拿的是老配置直到重启才生效。排查了大半天才找到根因。这个教训给我的启发是自定义迭代器的懒加载特性决定了它天然不适合用于那种“运行期间会变化的数据”作为缓存源。如果你确实需要缓存必须明确地设计失效机制——比如缓存里记录一个时间戳或者提供一个invalidata()方法让外部主动把缓存清空。迭代器里的缓存不是自动化的银弹它只是把“何时取数据”的决定权从调用方手里转移到了数据流内部这带来惰性优势的同时也把“数据新鲜度”责任抛给了实现者。6. 自定义迭代器在工程流水线中的实战价值6.1 为什么说它天然适合“懒处理”工程流水线里最常见的需求是从一个源头读取数据经过多个过滤、变换步骤最后输出到某个目的地。传统写法会把整个过程写成一个大函数里面穿插着for循环、if判断、中间列表最后塞进一个“主流程”。用一个例子说明。假设要处理一个很大的 CSV 文件里面有若干行每行包含日期、分类、数值三列。需求是解析每一行、过滤掉日期在 2023 年之前的行、将数值转为浮点数、按分类分组、统计每个分类的总和。非迭代器的写法import csv from collections import defaultdict with open(data.csv, r) as f: reader csv.reader(f) header next(reader) data [] for row in reader: date_str, category, amount row amount float(amount) if date_str 2023-01-01: data.append((date_str, category, amount)) result defaultdict(float) for date_str, category, amount in data: result[category] amount这段代码创建了中间列表data如果 CSV 文件有 100 万行这 100 万个元组全部驻留在内存里内存占用轻松上百 MB。迭代器流水线版import csv from collections import defaultdict def raw_rows(f): reader csv.reader(f) next(reader) # 跳过表头 for row in reader: yield row def filter_recent(rows): for row in rows: if row[0] 2023-01-01: yield row def to_typed(rows): for row in rows: yield row[0], row[1], float(row[2]) with open(data.csv, r) as f: rows raw_rows(f) rows filter_recent(rows) rows to_typed(rows) result defaultdict(float) for date_str, category, amount in rows: result[category] amount每个步骤都是一个生成器rows从源头一路传递到统计处数据是逐行流过的任何时刻内存里只有当前这一行和统计结果。你甚至可以把filter_recent和to_typed换成其他过滤器随意组合整个流水线的代码结构保持了线性、可读、可测试。这个风格的优点总结起来是三条一是内存占用与数据总量无关只与单条数据体积有关二是每一步都是独立的可复用组件可以单独单测三是流水线的组装和变更非常容易加一步就是再套一层生成器。6.2 一个完整的“多级流水线”构造实例我们扩展示例做一个更贴近实际的聚合计算从一个大的交易流水文件里按渠道分类统计各渠道在某个时间段内的总金额和交易笔数。假设每行数据格式为订单号, 渠道, 金额, 时间戳。分析需求只统计时间在 2023-01-01 到 2023-12-31 之间的交易按渠道汇总。import csv from collections import defaultdict def read_orders(file_obj): reader csv.DictReader(file_obj) for row in reader: yield row def in_year(row): ts int(row[timestamp]) if 1672531200 ts 1735689600: yield row def as_number(row): row[amount] float(row[amount]) yield row def aggregate(rows): summary defaultdict(lambda: {count: 0, total: 0.0}) for row in rows: channel row[channel] summary[channel][count] 1 summary[channel][total] row[amount] return summary with open(orders.csv, r) as f: rows read_orders(f) rows in_year(rows) rows as_number(rows) result aggregate(rows)在这个设计里in_year和as_number都是生成器并且都用yield逐条产出整个流水线在aggregate汇总之前不会产生任何中间大列表。aggregate不是生成器它消费完整条流水线后返回一个字典这是合理的结束动作。如果想扩展功能——比如再加一个“只保留金额大于 100 的订单”的过滤只需要插入一行def large_only(rows): for row in rows: if row[amount] 100: yield row rows large_only(rows)组件之间通过迭代器协议解耦新增处理步骤不用改动其他任何代码这就是流水线架构的扩展性。6.3 性能与内存实测我的量化和结论有些人可能会担心流水线模式一层一层包装生成器会不会比一口气读完再处理更慢我在自己的模拟数据上做过一个简单测试这里给出结果作为参考。我用一个约 50 万行、每行几十字节的 CSV 文件跑了两种方案。方案 A 是一次性readlines后逐行 parse方案 B 是上面的生成器流水线。结果内存峰值方案 A 约 320 MB方案 B 约 12 MB。差距接近 27 倍相当可观。总耗时方案 A 约 1.6 秒方案 B 约 2.1 秒。慢了约 30%但远没有到“不可接受”的程度。这个结果符合直觉生成器流水线因为每一层都要产生新的生成器对象、每次yield都有函数调用开销所以确实比一整块读入要慢一些。但在数据处理场景里内存往往比那几百毫秒更宝贵尤其是在服务器环境下内存占用过高可能导致 OOM甚至影响同台机器上的其他进程。如果确实对性能要求极高可以把多步生成器合并为一层减少 yield 次数。比如把in_year和as_number合并成一个filter_and_convert生成器用一次循环完成过滤和类型转换这样能减少一层函数调用。合并后我的测试里耗时可以降到 1.8 秒左右内存几乎没有变化。给我个人的结论是不带策略地优化不如先看数据和瓶颈在哪里。在内存不紧张、追求极限性能的场景用“一次读完”没问题但对绝大多数“在流上处理大文件”的业务迭代器流水线的内存收益是压倒性的慢那 30% 完全可以通过合并层数来弥补。7. 必须留意的坑从死循环到 StopIteration 被吞7.1 死循环无限迭代器忘了加终止条件前面提过无限序列这里单独拎出来强调一下它容易导致的故障。我自己就犯过这个错使用自定义迭代器的时候忘了next()调用在底层是依赖StopIteration来终止的一旦某个迭代器永远不抛这个异常for循环就像脱缰的野马永远不结束。比如下面这个迭代器本意是生产 0 到 9 的数class BrokenRange: def __init__(self, n): self.n n self.i 0 def __iter__(self): return self def __next__(self): if self.i self.n: v self.i self.i 1 return v raise StopIteration这段代码是对的。常见的错误是把raise StopIteration写成了return None或者写成了return StopIteration。前者一旦发生for循环会一直拿到None永远不终止——因为next()正常返回了None逻辑上这只是一个合法的元素。class AlsoBrokenRange: def __init__(self, n): self.n n self.i 0 def __iter__(self): return self def __next__(self): if self.i self.n: v self.i self.i 1 return v # 错误没有抛异常只是 return return这个类的for循环会输出 0 到 9 后继续无限输出None。排查这类问题最快的方法是打印前几个值看元素序列是否是预期的。如果你发现迭代到某个点之后不再前进、CPU 被打满优先怀疑StopIteration是否真的抛出了。7.2 顺手处理StopIteration的坑还有一个更隐蔽的坑发生在你手动调用next()时。如果我们自己捕获了StopIteration却不加区分可能把一个正常的“迭代结束”信号当成了错误吞掉。不少初学的人写try: item next(it) except StopIteration: item default这种写法本身没问题问题在于如果你在except块里又调用了另一个可能迭代结束的对象并且没有处理异常就会被外层吃掉。尤其当你写了一个函数内部同时在多个迭代器上做next()一旦某个迭代器的StopIteration没有被正确处理整个函数可能异常退出而调用方通过for循环感知到时它看到的只是“这个生成器提前结束了”根本不知道是哪个内部迭代器出了问题。我建议在所有内部逻辑跨多个迭代器的地方要么统一用for循环要么用itertools工具来协调不要手动在一大堆next()里到处 try。7.3 重新开始的困惑迭代器为何“用一次就没”“迭代器只能用一次”是最容易让人困惑的行为之一。拿文件对象举例with open(data.txt) as f: first_pass list(f) second_pass list(f) # 空的很多人第一次遇到会觉得奇怪明明文件还在啊为什么第二遍读不到东西原因就是文件对象本身就是迭代器内部维护了文件指针第一遍遍历已经把它推到了文件末尾。自定义迭代器同样如此。如果你实现了一个含状态的迭代器类务必在文档注释里写清楚这个迭代器是否是一次性的是否可以重复使用如何重新初始化如果这是一个对外提供的公共组件我建议提供明确的重置方法例如reset()或者rewind()内部把游标归零。如果需要完全可重复遍历更好的设计是把可迭代对象与迭代器分离每次迭代都创建一个全新的迭代器。7.4 多线程环境下的共享状态问题自定义迭代器如果被多个线程同时消费必须注意它的内部状态不是线程安全的。两个线程同时调用同一个迭代器的__next__可能出现竞态导致同一个元素被取两次或者游标错乱。实践中我有两种处理方式。一种很简单给__next__加一把锁变成线程安全的迭代器。另一种更常见放弃多线程共享同一个迭代器的想法每个线程都从同一个可迭代对象那里拿一个独立的迭代器实例。import threading class SafePagedAPI: def __init__(self, base_url): self.base_url base_url self._lock threading.Lock() def __iter__(self): return PagedAPIIterator(self.base_url) def __next__(self): with self._lock: return next(self._iter)第一个方案简单粗暴代价是每次next都要拿锁并发争抢会拖慢吞吐。第二个方案才更符合迭代器协议的本意——迭代器是轻量的、独立的遍历状态机共享状态本身就是一种错误的使用方式。8. 设计一个优秀自定义迭代器的五条经验法则8.1 法则一迭代器和可迭代对象分清楚这是最重要的设计决策。如果你做的是一个“数据源”希望调用方可以反复创建新的迭代器从零遍历就实现__iter__返回一个新的独立对象。如果你做的是一个“一次性消费的流”例如从网络读取数据、遍历一个实时传感器信号的光标直接让对象自己当迭代器即可。用个类比可迭代对象像自助餐厅你每次去调用iter都可以重新拿一个餐盘迭代器从头开始一排排取菜迭代器像电影放映机的胶片轴你把它从头放到底它就是一条单向不可逆的带子。8.2 法则二命名即接口自定义迭代器的类名和方法名要能让人一眼看出它的行为。例如PagedAPIIterator直接表明它是一个迭代器且源来自分页 APIBacktrackableIterator表明它支持回溯primes()表明它是一个素数生成器。避免仅仅叫ApiWrapper、DataStream这种看不出本质的名字。8.3 法则三越懒越好迭代器能“不做事”的地方尽量不做事。不要在__init__里去请求第一页数据不要在创建生成器对象时去读文件。把数据的获取、计算全部推迟到__next__或yield之后。这样创建迭代器的成本几乎为零真正的大开销发生在遍历的第一个元素被请求时。这个原则在调试时尤其好用。当你发现迭代器有问题你知道迭代器对象被创建时并没有执行真正的业务逻辑出错的点必然是next()被调用的时候排查范围缩小了很多。8.4 法则四显式地处理资源释放如果迭代器内部持有了文件句柄、网络连接、数据库游标等资源一定要设计清理机制。推荐用contextlib.closing或者让迭代器实现__enter__、__exit__方法让调用方可以配合with语句使用。生成器函数有一个值得知道的小技巧当生成器被垃圾回收或被close()时函数体内部的finally块会执行。你可以用这个机制自动释放资源def read_paged_api_segments(base_url): session requests.Session() try: page 0 while True: resp session.get(...) ... finally: session.close()这样即使生成器在中途被外部close()停止连接也会被安全关闭。8.5 法则五为调试留一扇窗迭代器是惰性求值的这意味着调试信息不那么直观。你无法在创建时打一个断点看到全部数据。实践中我会在迭代器里加一个可选的调试开关比如PagedAPI(base_url, debugTrue)在yield之前打一条带页码和元素序号的输出。这个小功能在排查“为什么遍历到一半数据不对”时帮了大忙。9. 具体场景下的取舍手写类 vs 生成器 vsitertools把前面的经验汇总一下我用一张表来总结不同实现方式的适用场景需求场景推荐方案原因简单线性遍历一个数据源生成器函数代码最短、最易读需要跨步骤共享复杂状态缓存、回退、控制逻辑手写类生成器在状态控制上力不从心多个迭代器组合、截断、分组聚合itertools配合生成器标准库已经写好组合能力极强需要线程安全地遍历共享数据源独立迭代器 容器模式避免共享状态自然线程安全需要回调、事件驱动式遍历手写类 回调注册生成器天然是拉模式不适合推模式需要反复从头遍历同一份内存数据可迭代对象 每次新建迭代器分离容器和迭代器举一个实际的组合案例。某业务需要把多个源文件的数据做一个 interleave 合并读取即第一行来自 A 文件第二行来自 B 文件第三行来自 A 文件……然后再做去重最后只保留符合条件的前 100 条import itertools def interleave(*file_iterators): round_robin itertools.cycle(file_iterators) done set() while len(done) len(file_iterators): for it in round_robin: try: line next(it) except StopIteration: if id(it) not in done: done.add(id(it)) continue yield line def dedup(iterable): seen set() for x in iterable: if x not in seen: seen.add(x) yield x def take_until_count(iterable, n): for i, x in enumerate(iterable): if i n: break yield x result list( take_until_count( dedup( interleave( open(a.txt), open(b.txt) ) ), 100 ) )这个例子里interleave是核心它处理多个迭代器“办不到”的特殊逻辑其余的都交给itertools和轻量生成器。当你对迭代器协议理解得够深你写出的数据处理代码就会越来越像管道每一层只管一小件事层与层之间通过“流”传递数据几乎不需要关心中间状态。写到这里我自己的体会是自定义迭代器设计不是一门“炫技”的技术它其实是对“数据从哪里来、到哪里去”这个问题的一种架构视角。当你习惯于用迭代器来表达数据处理流你会发现自己写出来的模块越来越薄、越来越容易测试、越来越愿意做懒加载。这种转变带来的收益往往比学会某一个具体 API 要大得多。在具体的项目里如果遇到了“从一个无法预知总量的数据源里逐个取元素”的需求或者“多步数据处理希望不占内存、可自由组合”的需求试着从迭代器协议入手去设计你大概率会得到一个比“一个巨大的函数”优雅得多的答案。最后分享一个实用小技巧。判断一个对象是不是迭代器最快的办法不是看类型而是看它能不能被next()直接调用。如果一个对象有__next__方法但iter(obj) is obj为 False说明它是一个可迭代对象而不是迭代器如果两者都是True它通常是一个“一次性迭代器”。可以用这段代码快速判断def is_iterator(obj): return hasattr(obj, __iter__) and hasattr(obj, __next__) and iter(obj) is obj这个判断方式我在代码评审里用了很长时间帮很多人快速理清了自己设计的数据结构到底属于“可以反复遍历的容器”还是“只能走一遍的流”。希望这篇内容也能帮你建立同样的直觉。 Python自定义迭代器完全指南：从协议到生成器与工程实战
1. 为什么自定义迭代器值得单独拿出来讲很多写了一段时间 Python 的人都有这种感觉列表、字典、集合这些内置类型用得很熟for循环也写得飞起但一旦提到“迭代器”三个字脑子里只剩一个模糊的印象——好像是个能next()的对象再问深一层for循环到底是怎么从列表里把元素一个个取出来的为什么有的对象可以for有的不行为什么有时候同一个对象能遍历两遍有时候却只能遍历一遍这些问题的答案都指向迭代器协议。我自己最初写业务代码的时候对迭代器的理解就是“能用for遍历的东西”完全没想过要自己去实现一个。直到有一次要处理一个很大的日志文件文件里每行是一个 JSON 对象总共几百万行。直接用readlines()读内存肯定爆一行行readline()写起来又啰嗦而且过滤器逻辑和文件读取逻辑混在一起代码很快就变得没法看了。当时我试着写了一个自定义迭代器类把“从文件里读一行 JSON”封装成__next__瞬间整个数据处理的代码变得清爽了而且内存占用稳定在一个很低的值。从那一刻起我才意识到自定义迭代器不是面试题里才用到的东西它是真实工程里非常有用的抽象工具。这篇博文不是讲 Python 内置函数iter()的简单用法而是聚焦在“如何从零设计一个自定义迭代器”。我会先拆解迭代器协议的本质然后从最朴素的类实现讲起再到生成器这种更高级的写法还会聊到双向迭代、无限序列、缓存机制、工程流水线里迭代器的实战价值最后把我踩过的一些坑和设计取舍一并交代清楚。适合的读者是那些已经会用 Python 写基本代码、但想进一步理解语言机制的人以及在实际项目中遇到过“遍历复杂数据源”“内存占用过高”“数据懒加载”等问题的开发者。2. 迭代器协议拆解__iter__和__next__到底在做什么2.1 一个对象能被for循环遍历本质是什么for循环是 Python 里最常见的遍历语法但它背后做事情的方式远比表面看起来更机械。当你写for item in container: print(item)Python 解释器实际上执行了这样几个步骤调用iter(container)尝试获取一个迭代器对象在一个隐藏的循环里反复调用next(iterator)每次拿到一个值就进入循环体直到next()抛出StopIteration异常循环便安静地结束。关键点在于for循环本身并不直接接触列表内部的数据。它接触的永远是“迭代器”。列表、元组、字典这类内置容器Python 会为它们各自生成一个独立的迭代器对象这个对象知道当前遍历到了哪个位置。列表和它的迭代器是两码事是很多新手容易忽略的一点。列表本身是容器它内部存了全部元素而迭代器是一个有状态的对象它只记录“下一个该取哪个元素”。举个例子lst [1, 2, 3] it iter(lst) print(next(it)) # 1 print(next(it)) # 2lst本身没有被消耗lst里的三个元素原封不动在那里。被消耗的是迭代器it它内部维护着一个游标取一次挪一格。取完第三个之后再调用next(it)就会抛StopIteration。2.2 协议对类的要求__iter__和__next__为了让一个自定义的类能被for循环消费它必须满足“迭代器协议”。协议在 Python 里不是强制的接口继承而是一种约定你只需要实现指定的方法解释器就会自动识别。核心方法有两个__iter__(self)返回一个迭代器对象。一个对象如果实现了这个方法它就是“可迭代的”iterable。注意这个方法返回的对象不一定就是self它可以是任何一个实现了__next__的对象。__next__(self)返回下一个元素。如果没有更多元素了必须抛出StopIteration异常。一个类如果同时实现了这两个方法它既是可迭代对象也是自身的迭代器。这种设计在简单场景下很好用但有一个隐蔽的坑同一个对象只能被完整遍历一次。因为迭代器的游标是对象自带的第一轮遍历结束后游标已经走到末尾第二轮直接就是空的。2.3 一个反直觉的机制序列协议的回退Python 判断一个对象是否可迭代并不是严格地看它有没有__iter__。如果对象实现了序列协议中的__getitem__也就是可以通过索引访问元素即使它没有__iter__iter()函数也会自动为它生成一个“索引型迭代器”。这是 Python 为了兼容老代码留下的一个后门机制。class SimpleRange: def __init__(self, start, end): self.start start self.end end def __getitem__(self, index): pos self.start index if pos self.end: raise IndexError return pos这个类没有__iter__也没有__next__但你可以直接for n in SimpleRange(1, 5)遍历它因为解释器发现它有__getitem__就从索引 0 开始依次调用遇到IndexError就认为遍历结束。这个机制本身很有用很多简单的数据结构只需要实现__getitem__就能获得遍历能力但它在排查问题时容易误导人。你往一个类里加了一个魔术方法结果发现它突然变得可以for了但明明没有实现迭代器协议——你得意识到是这个回退机制在起作用。3. 从零实现写一个真正能跑的自定义迭代器3.1 场景设定我要遍历的是什么自定义迭代器最常见的使用场景是遍历那些“不是天然躺在内存里的数据”或者“有多种遍历逻辑的数据”。光讲协议太抽象我们直接做一个有实际意义的例子模拟一个分页 API 的客户端迭代器。假设某外部数据接口采用分页方式返回数据请求 URL 形态为https://api.example.com/v1/pages?page1size100返回的 JSON 结构是{ data: [ {id: 1, name: A}, {id: 2, name: B} ], page: 1, has_more: true }如果你要在业务里把每一页的数据逐个拿出来处理最朴素的做法是写一个循环手动管理页码page 1 while True: resp requests.get(fhttps://api.example.com/v1/pages?page{page}size100) payload resp.json() for item in payload[data]: process(item) if not payload[has_more]: break page 1这段代码能跑但它把“请求网络”和“逐条处理”紧紧地耦合在了一个函数里。如果你有多个地方都要遍历这个接口的全部数据就得把这段逻辑复制粘贴若干次。而且page变量、while循环、break的判断全是过程式思维。换一个思路把它封装成一个迭代器遍历逻辑就变成了一件“关于数据源的事”处理逻辑只需要写一遍。3.2 朴素版实现状态机放在__next__里第一版不需要任何高级技巧直接在一个类里维护三个状态变量当前页码当前页内索引当前页的数据列表、是否还有更多页import requests class PagedAPI: def __init__(self, base_url, page_size100): self.base_url base_url self.page_size page_size self.current_page 0 self.items_in_page [] self.item_idx 0 self.has_more True def _fetch_page(self): resp requests.get( self.base_url, params{page: self.current_page, size: self.page_size} ) resp.raise_for_status() payload resp.json() self.items_in_page payload[data] self.has_more payload[has_more] def __iter__(self): return self def __next__(self): if self.item_idx len(self.items_in_page): if not self.has_more: raise StopIteration self.current_page 1 self._fetch_page() self.item_idx 0 if not self.items_in_page: raise StopIteration item self.items_in_page[self.item_idx] self.item_idx 1 return item使用方式变得非常简洁for item in PagedAPI(https://api.example.com/v1/pages, page_size100): process(item)写这个方法的时候有几点值得注意。__next__里必须先判断“当前页的索引是否已经越界”如果越界再判断“是否还有下一页”。如果当前页刚好取完最后一个元素item_idx等于列表长度这时候进入下一轮循环先看还有没有下一页有就拉取下一页并重置索引没有就直接抛StopIteration。这里有一个边界情况最后一页如果data是空的按照我的实现会触发if not self.items_in_page的兜底逻辑直接抛StopIteration结束遍历。这个迭代器把网络请求延迟到了真正需要数据的时候才发生这就是“懒加载”的核心思想。创建PagedAPI对象的时候不会发出任何 HTTP 请求只有当你开始for遍历元素不够用了它才去下一页拉数据。3.3 为什么__iter__返回self是有道理的上面代码里__iter__直接返回self意味着你把 PagedAPI 对象本身当成了自己的迭代器。这符合“迭代器是可迭代的”这一惯例但有个副作用同一个实例一旦被完整遍历过一遍就废了。api PagedAPI(https://api.example.com/v1/pages) list1 list(api) # 拉完了所有页 list2 list(api) # 直接就是 []如果你需要支持“可以多次遍历同一个数据源”更合适的设计是让__iter__每次返回一个新的迭代器对象而不是返回self。例如class PagedAPIContainer: def __init__(self, base_url): self.base_url base_url def __iter__(self): return PagedAPI(self.base_url)容器只有__iter__每次被for的时候都现场创建一个全新的PagedAPI迭代器遍历完就丢掉下次再遍历重新来一遍。这就是“可迭代对象”和“迭代器”各自的角色分工容器负责描述数据源迭代器负责记录遍历状态。这个设计取舍很重要。很多业务场景里“数据只能遍历一次”是可以接受的甚至就是期望的行为——比如流式处理一个大文件你并不想反复重读。但如果你要实现的是一个可以反复遍历的内存集合那就必须可控迭代器和容器分离。4. 生成器用yield把迭代器写得像同步代码4.1 生成器函数到底做了什么手动写__iter__和__next__的方法在简单场景下还行一旦状态转移比较复杂代码就变得有些冗长。比如上面那个分页 API 的迭代器真正核心的逻辑其实就两三行拉数据、逐条 yield。Python 提供了一种更符合直觉的写法——生成器函数。一个函数里只要出现yield关键字它就不再是普通函数而是生成器函数。调用它不会立刻执行函数体而是返回一个生成器对象。生成器对象是迭代器协议的内置实现你不需要手动维护游标状态解释器会在yield的位置帮你“挂起”和“恢复”。用生成器重写分页 APIdef iter_paged_api(base_url, page_size100): page 0 while True: resp requests.get( base_url, params{page: page, size: page_size} ) resp.raise_for_status() payload resp.json() for item in payload[data]: yield item if not payload[has_more]: break page 1这个函数读起来就像同步代码一样自然你写一个while循环拉数据yield一条拉完就break。生成器自动保存局部变量page自动暂停和恢复执行流你完全不需要关心item_idx之类的状态变量。对比一下生成器版本和类版本的差异非常明显对比维度手动类迭代器生成器函数状态管理需要自己维护变量解释器自动保存代码结构状态判断分散在多个方法里线性流程像普通函数可读性中等逻辑多时容易晕高数据流一目了然支持 send 传值需额外设计原生支持实现难度较高极低4.2 生成器不等于 itertools别混淆这里要澄清一个概念。很多教程会把itertools和生成器放在一起讲但它们是不同层面的东西。itertools是标准库里的一个工具模块里面提供的chain、groupby、islice、product这些函数绝大部分返回的是迭代器对象很多内部就是用 C 语言写的高效生成器。而生成器是你自己用yield写出来的自定义迭代逻辑。两者可以协同工作你可以把自己的生成器传给itertools的函数组合出更复杂的流水线。举个例子假如你想两页一取交错合并两个分页接口的数据import itertools api_a iter_paged_api(https://api.example.com/v1/a) api_b iter_paged_api(https://api.example.com/v1/b) for item_from_a, item_from_b in itertools.zip_longest(api_a, api_b): if item_from_a is not None: process(item_from_a) if item_from_b is not None: process(item_from_b)zip_longest会同时驱动两个迭代器直到两个都耗尽为止。两个生成器各自维护自己的分页状态互不干扰数据量大的场景下内存占用始终是常数级别的这种组合能力是手写类迭代器很难达到的。4.3 什么时候仍然选择手写类而不是生成器生成器虽然好用但并不是万能的。有几个场景我建议你回头用手写类第一当迭代器需要和外部系统交互、需要明确的close()清理资源时。生成器对象虽然有close()方法能在yield点抛出GeneratorExit但无论如何不如类里写一个__del__或显式的cleanup()来得清晰。第二当遍历逻辑需要频繁在“迭代”和其他操作之间切换时。生成器是一种“半协程”只能从上次yield的地方继续没法在迭代过程中从外部向内部发起额外的调用。如果你需要在遍历过程中提供“跳过”“回退”“暂停继续”等复杂控制手写类配合方法调用会更直观。第三性能极端敏感的场景。虽然是 CPython 里生成器的开销已经很小但每一次yield都是一次函数调用级别的成本在每秒几百万次迭代的核心循环里手写类配合局部变量优化可能更可控。不过说实话绝大多数业务场景都到不了这个量级优先考虑可读性就好。5. 进阶设计双向迭代、无限序列、缓存与失效5.1 双向迭代别硬套迭代器协议迭代器协议本质上只要求__next__返回下一个元素它天然是单向、一次性的。如果你需要从一个数据集合里既能往前又能往后有两条路可以走。第一条路无视迭代器协议直接设计一个独立的类提供next()和prev()方法让外部代码显式调用。这种设计最直观也最容易理解缺点是没法直接用for循环糖。第二条路在迭代器协议上叠加一个历史缓冲区记录已经遍历过的元素然后提供一个prev()方法让用户可以回到之前的位置。这个缓冲区可以是无限增长的列表也可以是有固定上限的滑动窗口。我给一个带回溯缓冲的迭代器设计思路class BacktrackableIterator: def __init__(self, iterable): self._src iter(iterable) self._buffer [] # 已经取出过的元素 self._cursor 0 # 当前游标指向下一个要返回的元素位置 def __iter__(self): return self def __next__(self): if self._cursor len(self._buffer): item self._buffer[self._cursor] self._cursor 1 return item item next(self._src) self._buffer.append(item) self._cursor 1 return item def prev(self): if self._cursor 0: raise IndexError(cannot go before the first item) self._cursor - 1 return self._buffer[self._cursor] def reset(self): self._cursor 0这个类的核心思路是底层源迭代器负责真正产生新元素已经产生过的元素全部暂存在_buffer里prev()只是把游标往回拨一下。如果往前拨到的位置已经出过元素就直接从缓冲区拿不会再向底层源要新数据。这种“基于迭代器 缓冲”的回溯方案在遍历流式数据时需要小心如果数据源是网络流或文件流底层迭代器推进之后就无法回头缓冲是实现回退的唯一手段所以缓冲区可能越来越大。解决方式是给缓冲区设置上限当回退深度超过上限时直接报错或提供截断策略。5.2 无限序列迭代器最惊艳的用法之一迭代器是唯一能在 Python 里优雅表示“无限序列”的方式因为它的计算是惰性的永远只计算下一个要的元素而不需要一次性把所有元素加载进内存。我来写一个素数生成器这是无限序列的经典案例def primes(): yield 2 seen [] candidate 3 while True: is_prime True limit int(candidate ** 0.5) 1 for p in seen: if p limit: break if candidate % p 0: is_prime False break if is_prime: seen.append(candidate) yield candidate candidate 2这个生成器可以一直迭代到任意大的素数但你不用担心内存爆炸——seen列表保存的是已经找到的素数用于后续整除判断空间复杂度是 O(已产生的素数个数)这是由“判断素数”这个需求本身决定的不是迭代器的锅。取前 10 个素数只需要import itertools first_ten list(itertools.islice(primes(), 10))islice从无限序列里截取前 10 个然后list把它耗尽。底层生成器的状态被驱动了 10 次之后islice会关闭它primes()内部循环自然就停了。无限序列的使用门槛在于你必须随时有“取够就停”的觉悟。写for p in primes():不带任何终止条件就是死循环。所以实践中无限迭代器场景常和itertools.islice、takewhile配合使用前者按数量截取后者按条件截取。5.3 缓存与失效什么时候迭代器让你“数据过期”自定义迭代器的优势是懒加载但懒加载有一个代价缓存的值可能过期。这里我想说一个真实踩过的坑能帮你避免类似的错误。某次我需要设计一个“按配置加载系统参数”的迭代器。配置本身是从数据库查的且业务运行期间可能有其他进程修改这份配置。我最初写了一个生成器在第一次迭代时从数据库把所有配置项查出来然后yield给调用方。这个生成器本身没有问题但问题出在调用方的使用姿势上——他们在模块加载时就调用了list(config_dict.items())把结果存到了一个全局变量里之后整个进程生命周期内用的都是这份缓存。结果是数据库里的配置改了但运行中的服务始终拿的是老配置直到重启才生效。排查了大半天才找到根因。这个教训给我的启发是自定义迭代器的懒加载特性决定了它天然不适合用于那种“运行期间会变化的数据”作为缓存源。如果你确实需要缓存必须明确地设计失效机制——比如缓存里记录一个时间戳或者提供一个invalidata()方法让外部主动把缓存清空。迭代器里的缓存不是自动化的银弹它只是把“何时取数据”的决定权从调用方手里转移到了数据流内部这带来惰性优势的同时也把“数据新鲜度”责任抛给了实现者。6. 自定义迭代器在工程流水线中的实战价值6.1 为什么说它天然适合“懒处理”工程流水线里最常见的需求是从一个源头读取数据经过多个过滤、变换步骤最后输出到某个目的地。传统写法会把整个过程写成一个大函数里面穿插着for循环、if判断、中间列表最后塞进一个“主流程”。用一个例子说明。假设要处理一个很大的 CSV 文件里面有若干行每行包含日期、分类、数值三列。需求是解析每一行、过滤掉日期在 2023 年之前的行、将数值转为浮点数、按分类分组、统计每个分类的总和。非迭代器的写法import csv from collections import defaultdict with open(data.csv, r) as f: reader csv.reader(f) header next(reader) data [] for row in reader: date_str, category, amount row amount float(amount) if date_str 2023-01-01: data.append((date_str, category, amount)) result defaultdict(float) for date_str, category, amount in data: result[category] amount这段代码创建了中间列表data如果 CSV 文件有 100 万行这 100 万个元组全部驻留在内存里内存占用轻松上百 MB。迭代器流水线版import csv from collections import defaultdict def raw_rows(f): reader csv.reader(f) next(reader) # 跳过表头 for row in reader: yield row def filter_recent(rows): for row in rows: if row[0] 2023-01-01: yield row def to_typed(rows): for row in rows: yield row[0], row[1], float(row[2]) with open(data.csv, r) as f: rows raw_rows(f) rows filter_recent(rows) rows to_typed(rows) result defaultdict(float) for date_str, category, amount in rows: result[category] amount每个步骤都是一个生成器rows从源头一路传递到统计处数据是逐行流过的任何时刻内存里只有当前这一行和统计结果。你甚至可以把filter_recent和to_typed换成其他过滤器随意组合整个流水线的代码结构保持了线性、可读、可测试。这个风格的优点总结起来是三条一是内存占用与数据总量无关只与单条数据体积有关二是每一步都是独立的可复用组件可以单独单测三是流水线的组装和变更非常容易加一步就是再套一层生成器。6.2 一个完整的“多级流水线”构造实例我们扩展示例做一个更贴近实际的聚合计算从一个大的交易流水文件里按渠道分类统计各渠道在某个时间段内的总金额和交易笔数。假设每行数据格式为订单号, 渠道, 金额, 时间戳。分析需求只统计时间在 2023-01-01 到 2023-12-31 之间的交易按渠道汇总。import csv from collections import defaultdict def read_orders(file_obj): reader csv.DictReader(file_obj) for row in reader: yield row def in_year(row): ts int(row[timestamp]) if 1672531200 ts 1735689600: yield row def as_number(row): row[amount] float(row[amount]) yield row def aggregate(rows): summary defaultdict(lambda: {count: 0, total: 0.0}) for row in rows: channel row[channel] summary[channel][count] 1 summary[channel][total] row[amount] return summary with open(orders.csv, r) as f: rows read_orders(f) rows in_year(rows) rows as_number(rows) result aggregate(rows)在这个设计里in_year和as_number都是生成器并且都用yield逐条产出整个流水线在aggregate汇总之前不会产生任何中间大列表。aggregate不是生成器它消费完整条流水线后返回一个字典这是合理的结束动作。如果想扩展功能——比如再加一个“只保留金额大于 100 的订单”的过滤只需要插入一行def large_only(rows): for row in rows: if row[amount] 100: yield row rows large_only(rows)组件之间通过迭代器协议解耦新增处理步骤不用改动其他任何代码这就是流水线架构的扩展性。6.3 性能与内存实测我的量化和结论有些人可能会担心流水线模式一层一层包装生成器会不会比一口气读完再处理更慢我在自己的模拟数据上做过一个简单测试这里给出结果作为参考。我用一个约 50 万行、每行几十字节的 CSV 文件跑了两种方案。方案 A 是一次性readlines后逐行 parse方案 B 是上面的生成器流水线。结果内存峰值方案 A 约 320 MB方案 B 约 12 MB。差距接近 27 倍相当可观。总耗时方案 A 约 1.6 秒方案 B 约 2.1 秒。慢了约 30%但远没有到“不可接受”的程度。这个结果符合直觉生成器流水线因为每一层都要产生新的生成器对象、每次yield都有函数调用开销所以确实比一整块读入要慢一些。但在数据处理场景里内存往往比那几百毫秒更宝贵尤其是在服务器环境下内存占用过高可能导致 OOM甚至影响同台机器上的其他进程。如果确实对性能要求极高可以把多步生成器合并为一层减少 yield 次数。比如把in_year和as_number合并成一个filter_and_convert生成器用一次循环完成过滤和类型转换这样能减少一层函数调用。合并后我的测试里耗时可以降到 1.8 秒左右内存几乎没有变化。给我个人的结论是不带策略地优化不如先看数据和瓶颈在哪里。在内存不紧张、追求极限性能的场景用“一次读完”没问题但对绝大多数“在流上处理大文件”的业务迭代器流水线的内存收益是压倒性的慢那 30% 完全可以通过合并层数来弥补。7. 必须留意的坑从死循环到 StopIteration 被吞7.1 死循环无限迭代器忘了加终止条件前面提过无限序列这里单独拎出来强调一下它容易导致的故障。我自己就犯过这个错使用自定义迭代器的时候忘了next()调用在底层是依赖StopIteration来终止的一旦某个迭代器永远不抛这个异常for循环就像脱缰的野马永远不结束。比如下面这个迭代器本意是生产 0 到 9 的数class BrokenRange: def __init__(self, n): self.n n self.i 0 def __iter__(self): return self def __next__(self): if self.i self.n: v self.i self.i 1 return v raise StopIteration这段代码是对的。常见的错误是把raise StopIteration写成了return None或者写成了return StopIteration。前者一旦发生for循环会一直拿到None永远不终止——因为next()正常返回了None逻辑上这只是一个合法的元素。class AlsoBrokenRange: def __init__(self, n): self.n n self.i 0 def __iter__(self): return self def __next__(self): if self.i self.n: v self.i self.i 1 return v # 错误没有抛异常只是 return return这个类的for循环会输出 0 到 9 后继续无限输出None。排查这类问题最快的方法是打印前几个值看元素序列是否是预期的。如果你发现迭代到某个点之后不再前进、CPU 被打满优先怀疑StopIteration是否真的抛出了。7.2 顺手处理StopIteration的坑还有一个更隐蔽的坑发生在你手动调用next()时。如果我们自己捕获了StopIteration却不加区分可能把一个正常的“迭代结束”信号当成了错误吞掉。不少初学的人写try: item next(it) except StopIteration: item default这种写法本身没问题问题在于如果你在except块里又调用了另一个可能迭代结束的对象并且没有处理异常就会被外层吃掉。尤其当你写了一个函数内部同时在多个迭代器上做next()一旦某个迭代器的StopIteration没有被正确处理整个函数可能异常退出而调用方通过for循环感知到时它看到的只是“这个生成器提前结束了”根本不知道是哪个内部迭代器出了问题。我建议在所有内部逻辑跨多个迭代器的地方要么统一用for循环要么用itertools工具来协调不要手动在一大堆next()里到处 try。7.3 重新开始的困惑迭代器为何“用一次就没”“迭代器只能用一次”是最容易让人困惑的行为之一。拿文件对象举例with open(data.txt) as f: first_pass list(f) second_pass list(f) # 空的很多人第一次遇到会觉得奇怪明明文件还在啊为什么第二遍读不到东西原因就是文件对象本身就是迭代器内部维护了文件指针第一遍遍历已经把它推到了文件末尾。自定义迭代器同样如此。如果你实现了一个含状态的迭代器类务必在文档注释里写清楚这个迭代器是否是一次性的是否可以重复使用如何重新初始化如果这是一个对外提供的公共组件我建议提供明确的重置方法例如reset()或者rewind()内部把游标归零。如果需要完全可重复遍历更好的设计是把可迭代对象与迭代器分离每次迭代都创建一个全新的迭代器。7.4 多线程环境下的共享状态问题自定义迭代器如果被多个线程同时消费必须注意它的内部状态不是线程安全的。两个线程同时调用同一个迭代器的__next__可能出现竞态导致同一个元素被取两次或者游标错乱。实践中我有两种处理方式。一种很简单给__next__加一把锁变成线程安全的迭代器。另一种更常见放弃多线程共享同一个迭代器的想法每个线程都从同一个可迭代对象那里拿一个独立的迭代器实例。import threading class SafePagedAPI: def __init__(self, base_url): self.base_url base_url self._lock threading.Lock() def __iter__(self): return PagedAPIIterator(self.base_url) def __next__(self): with self._lock: return next(self._iter)第一个方案简单粗暴代价是每次next都要拿锁并发争抢会拖慢吞吐。第二个方案才更符合迭代器协议的本意——迭代器是轻量的、独立的遍历状态机共享状态本身就是一种错误的使用方式。8. 设计一个优秀自定义迭代器的五条经验法则8.1 法则一迭代器和可迭代对象分清楚这是最重要的设计决策。如果你做的是一个“数据源”希望调用方可以反复创建新的迭代器从零遍历就实现__iter__返回一个新的独立对象。如果你做的是一个“一次性消费的流”例如从网络读取数据、遍历一个实时传感器信号的光标直接让对象自己当迭代器即可。用个类比可迭代对象像自助餐厅你每次去调用iter都可以重新拿一个餐盘迭代器从头开始一排排取菜迭代器像电影放映机的胶片轴你把它从头放到底它就是一条单向不可逆的带子。8.2 法则二命名即接口自定义迭代器的类名和方法名要能让人一眼看出它的行为。例如PagedAPIIterator直接表明它是一个迭代器且源来自分页 APIBacktrackableIterator表明它支持回溯primes()表明它是一个素数生成器。避免仅仅叫ApiWrapper、DataStream这种看不出本质的名字。8.3 法则三越懒越好迭代器能“不做事”的地方尽量不做事。不要在__init__里去请求第一页数据不要在创建生成器对象时去读文件。把数据的获取、计算全部推迟到__next__或yield之后。这样创建迭代器的成本几乎为零真正的大开销发生在遍历的第一个元素被请求时。这个原则在调试时尤其好用。当你发现迭代器有问题你知道迭代器对象被创建时并没有执行真正的业务逻辑出错的点必然是next()被调用的时候排查范围缩小了很多。8.4 法则四显式地处理资源释放如果迭代器内部持有了文件句柄、网络连接、数据库游标等资源一定要设计清理机制。推荐用contextlib.closing或者让迭代器实现__enter__、__exit__方法让调用方可以配合with语句使用。生成器函数有一个值得知道的小技巧当生成器被垃圾回收或被close()时函数体内部的finally块会执行。你可以用这个机制自动释放资源def read_paged_api_segments(base_url): session requests.Session() try: page 0 while True: resp session.get(...) ... finally: session.close()这样即使生成器在中途被外部close()停止连接也会被安全关闭。8.5 法则五为调试留一扇窗迭代器是惰性求值的这意味着调试信息不那么直观。你无法在创建时打一个断点看到全部数据。实践中我会在迭代器里加一个可选的调试开关比如PagedAPI(base_url, debugTrue)在yield之前打一条带页码和元素序号的输出。这个小功能在排查“为什么遍历到一半数据不对”时帮了大忙。9. 具体场景下的取舍手写类 vs 生成器 vsitertools把前面的经验汇总一下我用一张表来总结不同实现方式的适用场景需求场景推荐方案原因简单线性遍历一个数据源生成器函数代码最短、最易读需要跨步骤共享复杂状态缓存、回退、控制逻辑手写类生成器在状态控制上力不从心多个迭代器组合、截断、分组聚合itertools配合生成器标准库已经写好组合能力极强需要线程安全地遍历共享数据源独立迭代器 容器模式避免共享状态自然线程安全需要回调、事件驱动式遍历手写类 回调注册生成器天然是拉模式不适合推模式需要反复从头遍历同一份内存数据可迭代对象 每次新建迭代器分离容器和迭代器举一个实际的组合案例。某业务需要把多个源文件的数据做一个 interleave 合并读取即第一行来自 A 文件第二行来自 B 文件第三行来自 A 文件……然后再做去重最后只保留符合条件的前 100 条import itertools def interleave(*file_iterators): round_robin itertools.cycle(file_iterators) done set() while len(done) len(file_iterators): for it in round_robin: try: line next(it) except StopIteration: if id(it) not in done: done.add(id(it)) continue yield line def dedup(iterable): seen set() for x in iterable: if x not in seen: seen.add(x) yield x def take_until_count(iterable, n): for i, x in enumerate(iterable): if i n: break yield x result list( take_until_count( dedup( interleave( open(a.txt), open(b.txt) ) ), 100 ) )这个例子里interleave是核心它处理多个迭代器“办不到”的特殊逻辑其余的都交给itertools和轻量生成器。当你对迭代器协议理解得够深你写出的数据处理代码就会越来越像管道每一层只管一小件事层与层之间通过“流”传递数据几乎不需要关心中间状态。写到这里我自己的体会是自定义迭代器设计不是一门“炫技”的技术它其实是对“数据从哪里来、到哪里去”这个问题的一种架构视角。当你习惯于用迭代器来表达数据处理流你会发现自己写出来的模块越来越薄、越来越容易测试、越来越愿意做懒加载。这种转变带来的收益往往比学会某一个具体 API 要大得多。在具体的项目里如果遇到了“从一个无法预知总量的数据源里逐个取元素”的需求或者“多步数据处理希望不占内存、可自由组合”的需求试着从迭代器协议入手去设计你大概率会得到一个比“一个巨大的函数”优雅得多的答案。最后分享一个实用小技巧。判断一个对象是不是迭代器最快的办法不是看类型而是看它能不能被next()直接调用。如果一个对象有__next__方法但iter(obj) is obj为 False说明它是一个可迭代对象而不是迭代器如果两者都是True它通常是一个“一次性迭代器”。可以用这段代码快速判断def is_iterator(obj): return hasattr(obj, __iter__) and hasattr(obj, __next__) and iter(obj) is obj这个判断方式我在代码评审里用了很长时间帮很多人快速理清了自己设计的数据结构到底属于“可以反复遍历的容器”还是“只能走一遍的流”。希望这篇内容也能帮你建立同样的直觉。