1. 为什么我建议你重新认识一下迭代器如果你写过一阵Python大概率用过for循环、列表推导式、map、filter这些写法但你有没有想过这样一个问题我们平时说的“遍历一个列表”底层到底发生了什么为什么列表能被for循环读取而一个自定义的类实例默认情况下却不行答案就是迭代器协议。这两个方法——__iter__和__next__——决定了某个对象能不能被遍历以及在遍历时按什么规则产出数据。网上讲迭代器的文章非常多但大部分停留在“迭代器是什么”的层面真正把“如何自己设计一个迭代器”讲透的很少。这篇博文不打算重复教科书上的定义我会直接从一个实际场景出发带你一步步把自定义迭代器的设计思路、实现路径和常见坑全部过一遍。文章适合下面几类读者已经会用Python写基础程序但想搞清楚for循环背后运行机制的初学者需要处理大文件、数据流、无限序列等场景但不知道如何高效设计遍历逻辑的中级开发者想在设计类库或框架时提供更优雅的遍历接口让调用方像使用原生容器一样使用你的对象的高级开发者。我默认你掌握了Python的基础语法类、方法、yield关键字至少见过但即使你对yield只停留在“好像能生成序列”的层面也没关系下文会从原理层面讲到它和迭代器的关系。2. 迭代器协议的核心机制两个方法如何让对象“可循环”2.1 拆解for循环的底层动作先来看一个最基础的问题当Python执行for item in obj时它到底执行了哪些步骤实际上解释器做了这样几件事调用iter(obj)尝试获取一个迭代器对象反复调用这个迭代器对象的__next__()方法每次拿到一个值当__next__()抛出StopIteration异常时循环正常结束。也就是说for循环本质上是一个“反复调用next()直到异常”的语法糖。验证一下。我们用iter()取一个列表的迭代器然后手动调用next()lst [10, 20, 30] it iter(lst) print(next(it)) # 10 print(next(it)) # 20 print(next(it)) # 30 print(next(it)) # 抛出 StopIteration这里的关键在于iter(lst)返回的不是列表本身而是一个列表迭代器list_iterator。它内部记录着“当前读到了哪个位置”每次next()推进一位。搞清楚这一点之后“自定义迭代器”就有了清晰的蓝图你要做的就是让某个对象支持__iter__和__next__这两个方法或者用一种更简便的方式后面会讲到的生成器让Python自动帮你实现这两个方法。2.2iter和next的分工在Python的迭代协议里两个方法各司其职__iter__(self)返回一个迭代器对象。这个对象必须自己实现__next__方法。注意返回的不一定是self本身完全可以返回一个新的、独立的迭代器对象。__next__(self)每一次调用应该返回序列中的下一个值如果没有更多值了必须抛出StopIteration。下图可以帮助你理解这两者的关系这里用文字描述一下逻辑链路可迭代对象实现了 __iter__ → iter() 调用 __iter__ 得到 迭代器实现了 __next__ → next() 调用 __next__ 取值 → 没有更多值时抛出 StopIteration → for 循环结束有一点我建议你记牢可迭代对象和迭代器不是一回事。列表是可迭代对象但它本身不是迭代器——因为列表没有__next__方法。每次for遍历列表时iter()都会给出一个新的迭代器所以列表可以被循环遍历很多次而迭代器是有状态的它内部记录了位置遍历一遍之后就用完了无法从头再来。2.3 一个最小可用的自定义迭代器基于上面的协议写一个最简单的自定义迭代器。比如我想实现一个“倒计时”迭代器每次产生一个数字从3递减到0class 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 num in Countdown(3): print(num)运行结果3 2 1 0这就是自定义迭代器最基础的样子。__iter__返回self因为Countdown实例本身就可以充当自己的迭代器__next__维护current状态当数字递减到-1时抛出StopIteration通知循环结束。从这个例子里可以提炼出设计迭代器的三个要点必须有地方记录“当前遍历到了哪一步”状态变量每次next调用时要决定“下一步该返回什么”遍历结束时必须清晰地抛出StopIteration——这是协议的一部分漏掉的话for循环会一直执行下去。3. 两种典型实现路径类实现与生成器实现3.1 类实现状态具象化适合复杂逻辑刚才的Countdown是类实现的典型样板。类实现的好处在于状态是显式的属性比如self.current你可以随时查看、修改甚至持久化它适合承载复杂逻辑。比如你想在迭代过程中动态改变行为、记录统计信息、或维护多个状态变量类方案更直观。再举一个更实际的例子。假设你要实现一个迭代器用于逐行读取一个大文件但同时要统计已读取的总字符数和行数。如果只用生成器统计逻辑需要额外包装用类实现这些信息就是天然的对象属性class FileLineStats: def __init__(self, file_path, encodingutf-8): self.file_path file_path self.encoding encoding self._file None self.line_count 0 self.char_count 0 def __iter__(self): self._file open(self.file_path, encodingself.encoding) self.line_count 0 self.char_count 0 return self def __next__(self): line self._file.readline() if not line: # readline 在文件末尾返回空字符串 self._file.close() raise StopIteration self.line_count 1 self.char_count len(line) return line.rstrip(\n)注意这里的__iter__写法每次迭代开始时重新打开文件、清零计数器这样同一个FileLineStats对象可以被for循环多次使用而且每次都是从头开始。这一点很关键后文避坑部分还会展开。3.2 生成器实现极简语法80%场景的首选生成器generator是Python实现迭代器的语法糖。任何包含yield关键字的函数调用后返回的都是一个生成器对象而这个生成器对象天然满足迭代器协议。用生成器重写倒计时def countdown(start): while start 0: yield start start - 1就这么简单。没有__iter__、没有__next__、没有StopIteration——生成器内部自动处理了这一切。调用countdown(3)时不会执行函数体而是返回一个生成器对象当你for num in countdown(3)时生成器开始执行每次遇到yield就把值抛出来然后暂停下次继续。生成器最强大的地方在于惰性求值它不会一次性在内存中生成整个序列而是每次调用next()时才计算下一个值。这意味着理论上你可以创建一个无限序列的迭代器而不占用无穷内存这一点在后面的实战案例中会体现。3.3 两种方案怎么选我把两种实现方式的特性整理成一个表格方便你对照决策维度类实现生成器实现代码量较多需要显式写__iter__和__next__极少核心逻辑只需yield状态管理显式属性可随时访问和修改状态躲在函数局部变量里由解释器管理复杂逻辑承载好可在类里加辅助方法一般逻辑复杂时函数体会变乱多次迭代可控可设计为每次迭代返回新的状态生成器对象一次性消耗重新遍历需重新调用函数调试体验属性可查调试友好断点停住时局部变量也能看但状态流转不够直观我的个人经验是80%的迭代器需求用生成器就够了。尤其是数据流处理、管道式转换、无限序列等场景生成器的惰性求值几乎是量身定做。类实现更多用在需要伴随“辅助接口”的场景比如迭代器自身还要提供peek()、reset()、统计信息等方法时用类把状态和动作封装在一起更合理。4. 真实业务场景下的迭代器设计实战4.1 场景一无限序列的惰性生成这个场景几乎是生成器的主场。比如生成无限斐波那契数列def fibonacci(): a, b 0, 1 while True: yield a a, b b, a b调用方式fib fibonacci() for _ in range(10): print(next(fib), end ) # 输出0 1 1 2 3 5 8 13 21 34注意这个迭代器是“无底洞”它永远不会抛出StopIteration。这正是它的价值所在你可以按需取用任意多个斐波那契数而不必事先确定序列长度。消费方用多少次next()它就算多少次之后的计算完全延后。在这个基础上可以封装出很多实用的小工具。比如取出前N项from itertools import islice fib fibonacci() first_20 list(islice(fib, 20))又比如配合条件过滤找到第一个满足条件的斐波那契数fib fibonacci() first_over_1000 next(num for num in fib if num 1000)这些都是生成器与标准库itertools配合的典型操作。如果你在设计一个需要按需生成大量数据的API这种“取用方控制节奏”的设计思路值得借鉴。4.2 场景二大文件流式读取的分块迭代处理大文件时一次性read()会把整个文件加载进内存文件稍大一点就容易内存溢出。比较常规的做法是逐行readline()但如果你要读取的是二进制文件、或者需要按固定字节数分块读取呢这时自定义迭代器就很香了。下面实现一个按指定块大小读取文件的迭代器def read_chunks(file_path, chunk_size8192): with open(file_path, rb) as f: while True: chunk f.read(chunk_size) if not chunk: break yield chunk这是一个经典的“分块读取”工具。f.read(chunk_size)每次最多读取chunk_size字节文件读完时返回空字节串b此时break结束循环。实际使用中你可以把它跟文件解析逻辑组合起来边读边处理for chunk in read_chunks(large_video.mp4, 1024 * 512): # 在这里对 chunk 做处理比如计算哈希、上传到远端等 pass相比一次性读取整个文件这种写法的内存占用是恒定的只和chunk_size有关。对于几百兆甚至几个G的文件这种设计几乎是唯一实用的方案。这里有一个细节要注意如果你用for循环配合read_chunks循环结束后文件会由with块自动关闭不需要手动去关。这就是生成器配合上下文管理器的好处——资源清理的逻辑集中在函数里调用方完全不用操心。4.3 场景三树形结构的深度优先遍历迭代器再来一个稍微复杂一点的场景。假设你有一个多叉树节点结构如下class TreeNode: def __init__(self, value): self.value value self.children []你想实现一个“深度优先遍历”的迭代器让调用方可以直接for node in dfs_traverse(root): print(node.value)如果不用迭代器你可能会写一个递归函数把所有值收集到列表里再返回def dfs_values(node, result): result.append(node.value) for child in node.children: dfs_values(child, result) return result这种方法在树比较深、节点比较多时会一次性创建一个大列表而且递归实现本身有栈溢出的风险。用生成器改写逻辑会精简很多def dfs_traverse(node): yield node.value for child in node.children: yield from dfs_traverse(child)注意这里的yield from——它的作用是“将子生成器产出的每一个值转交给外层生成器”。换句话说yield from相当于在当前生成器内部嵌套遍历另一个迭代器并把它的产出值逐个抛给调用方。这个语法是从Python 3.3开始支持的在处理嵌套迭代时非常方便。如果你希望遍历时顺便记录路径或层次信息也很容易扩展def dfs_with_depth(node, depth0): yield node.value, depth for child in node.children: yield from dfs_with_depth(child, depth 1)调用时for value, depth in dfs_with_depth(root): print( * depth str(value))这样就能按缩进直观地展示树的层次结构。4.4 场景四迭代过程中动态追加任务这个场景是我在实际工作中踩过坑后封装出来的。设想你有一个待处理的任务列表处理过程中可能会产生新的子任务需要继续加入队列。如果一开始用普通列表配合for循环你会发现for循环遍历列表时如果你在循环体内append新元素新元素后续还是会被遍历到但一旦用索引遍历就容易出各种边界问题。更干净的做法是用迭代器模拟一个可增长的队列def task_processor(initial_tasks): tasks list(initial_tasks) index 0 while index len(tasks): task tasks[index] index 1 # 处理当前任务并可能产生新任务 new_tasks process_task(task) tasks.extend(new_tasks) yield task调用方只需要for task in task_processor([start]): print(处理:, task)在这里迭代器内部维护了一个不断增长的列表while循环的条件每次都会重新检查len(tasks)所以动态追加的任务也会被依次处理直到队列清空。这种“工作队列”模式在很多场景下都能派上用场比如遍历网站链接时发现新链接、遍历目录时发现子目录等。5. 设计迭代器时必须避开的几个坑5.1 迭代器的“一次性”到底意味着什么这是新手最容易踩的坑。先看一个现象values [1, 2, 3] it iter(values) print(list(it)) # [1, 2, 3] print(list(it)) # []第二次调用list(it)得到空列表因为第一次调用已经把it消费完了。迭代器的这种“单向流动”特性有时候会带来隐蔽的bug。比如def get_even_numbers(numbers): return (n for n in numbers if n % 2 0) gen get_even_numbers([1, 2, 3, 4, 5, 6]) if 2 in gen: # 这里会消耗掉 2 之前的生成器状态 print(找到了2) print(list(gen)) # 只会输出 [4, 6]而不是 [2, 4, 6]in关键字会逐个迭代生成器直到找到目标。这个过程是有副作用的它会在找到目标后停下但生成器已经前进到了那个位置。之后再list(gen)只会得到剩余部分。我在设计迭代器时经常会先想清楚一个问题这个迭代器被消费一次之后是否需要支持再次从头遍历如果需要就要么让__iter__每次返回新的迭代器对象要么在文档里明确提示调用方“每次遍历需要重新调用工厂方法”。比如前面的FileLineStats我刻意让__iter__重置状态就是为了同一个对象可以被重复for。5.2 惰性求值的双刃剑外部状态污染生成器的惰性求值虽然省内存但也意味着它在执行时依赖的外部状态可能已经变化。看这个例子def generate_from_list(lst): for item in lst: yield item data [1, 2, 3] gen generate_from_list(data) data.append(4) # 在生成器还没有遍历完的时候修改数据源 print(list(gen)) # [1, 2, 3, 4]生成器会“看见”追加的新元素这到底是好事还是坏事取决于你的设计意图。如果你明确知道迭代是一个“快照”操作那么这个行为就出乎意料如果你希望迭代器实时反映数据源的变化那反而是优点。我的建议是在自定义迭代器时如果数据源可能在迭代过程中被动修改最好在文档或类型注释里说明这一行为。如果希望迭代器不受外部影响可以在构造迭代器时主动拷贝一份数据class SnapshotIterator: def __init__(self, data): self._data list(data) # 构造时拷贝迭代期间不受外部修改影响 self._index 0 def __iter__(self): return self def __next__(self): if self._index len(self._data): raise StopIteration value self._data[self._index] self._index 1 return value5.3 迭代过程中修改容器的风险Python在遍历dict、list等内置容器时如果同时在循环体内修改容器会直接抛出RuntimeErrord {a: 1, b: 2} for key in d: if key a: del d[key] # RuntimeError: dictionary changed size during iteration这是我见过不少人踩过的坑。“遍历时不能修改容器”这条规则在自定义迭代器里也值得注意。如果你的迭代器内部依赖某个列表或字典并且在__next__中会改动它那就要想清楚会不会带来不可预期的行为。比较安全的做法是把“需要删除/修改的元素”先收集起来遍历结束后统一处理def filter_items(mapping): keys_to_delete [] for key, value in mapping.items(): if value 0: keys_to_delete.append(key) for key in keys_to_delete: del mapping[key]或者更简洁地用一个生成器来产出符合条件的键让调用方决定是否修改原容器def keys_with_positive_value(mapping): for key, value in mapping.items(): if value 0: yield key把“遍历”和“修改”分开思路会清晰很多。5.4 嵌套迭代器共享状态导致的诡异行为当你把两个迭代器组合使用时如果它们内部不小心共享了可变状态就会出现难以排查的bug。比如假设你写了一个迭代器类内部使用类属性保存状态class BadIterator: current 0 # 类属性所有实例共享 def __iter__(self): return self def __next__(self): self.current 1 return self.current然后创建两个实例同时遍历a BadIterator() b BadIterator() print(next(a)) # 1 print(next(b)) # 2而不是 1原因是current是类属性所有实例共享同一份状态。这个问题在类实现迭代器时尤其容易犯。解决办法很简单状态一定存放在实例属性里比如在__init__中self.current 0。这个案例提醒我设计迭代器时状态的“归属”要想清楚。谁拥有状态谁就应该负责状态的初始化、更新和清理。如果状态被多个迭代器实例共享多半会出现并发环境下的竞争问题——即使不是并发环境也容易互相干扰。5.5 无限迭代器与终止条件的边界管理无限迭代器非常方便但如果使用它的代码逻辑不完整很容易造成死循环。典型场景是def count_from(n): while True: yield n n 1 for num in count_from(10): if num 100: break print(num)这段代码有正确的退出条件所以没问题。但如果你忘了写break程序就会永远跑下去。因此我在设计无限迭代器时往往会额外提供一个“有界版本”的工厂函数def count_from(n, limitNone): while limit is None or n limit: yield n n 1这样调用方可以根据自己的需求选择使用有界版本还是无限版本。如果你的迭代器是对外暴露的API这种“提供边界控制的选项”会显著降低使用者的心智负担。6. 迭代器与生成器的进阶技巧6.1 用yield from简化嵌套迭代前面提到yield from时简单提了一句这里再展开说一下。yield from的作用是“代理迭代”等价于下面这种写法def flatten(nested): for sublist in nested: for item in sublist: yield item # 等价写法 def flatten(nested): for sublist in nested: yield from sublist两者的输出完全一致nested [[1, 2], [3, 4], [5]] print(list(flatten(nested))) # [1, 2, 3, 4, 5]在自定义迭代器中yield from能显著减少嵌套循环的层级。比如你要实现一个“遍历多层目录下所有文件”的迭代器import os def all_files(root_dir): for entry in os.scandir(root_dir): if entry.is_file(): yield entry.path elif entry.is_dir(): yield from all_files(entry.path)这段代码用递归加yield from实现了深度优先的目录遍历代码量极少逻辑却非常清晰。6.2 为迭代器增加send、throw和close如果你觉得生成器只能“单向产出”数据那就错了。Python的生成器还有send(value)方法可以向生成器内部“发送”一个值这个值会成为当前暂停的yield表达式的返回值。看个例子def accumulator(): total 0 while True: value yield total if value is None: continue total value gen accumulator() print(next(gen)) # 0启动生成器并执行到 yield total print(gen.send(10)) # 10total 变为 10 print(gen.send(5)) # 15total 变为 15send让生成器具备了“外来输入”的能力这在实现协程、状态机等场景时非常有用。不过在自定义迭代器中绝大多数场景不需要send我也建议你先从普通的yield开始确实有双向通信需求时再引入send。除了send还有throw()和close()两个方法。throw()可以在生成器暂停处抛入一个异常close()可以终止生成器。这些方法日常用得不多但当你设计需要提前终止清理资源的生成器时close()配合try/finally能帮你做好收尾工作def resource_generator(): try: print(资源打开) for i in range(10): yield i finally: print(资源释放) gen resource_generator() print(next(gen)) gen.close() # 输出 # 资源打开 # 0 # 资源释放6.3 与itertools标准库的组合使用itertools是Python迭代器生态中不可或缺的一部分自定义迭代器和它配合能大幅提升表达能力。常用组合包括itertools.islice对无限迭代器做切片取出前N个元素itertools.takewhile直到某个条件不满足为止持续取值itertools.chain把多个迭代器拼接成一个itertools.zip_longest多个迭代器并行遍历短的用填充值补齐。举个例子用无限斐波那契和takewhile结合取出所有小于1000的斐波那契数from itertools import takewhile fib fibonacci() result list(takewhile(lambda x: x 1000, fib)) # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987]这里takewhile会自动在某个值不满足条件时停止迭代相当于给无限迭代器加了一个动态边界。这种组合的思路是自定义迭代器负责定义数据产生规则itertools负责定义消费策略两者解耦各司其职。7. 设计迭代器的几个习惯性思考讲完了实现路径、实战案例和避坑如果从这些经验里提炼几条可以复用的“设计习惯”大概是下面这些。先想清楚“谁拥有状态”。迭代器是一种有状态的对象而状态的所有者决定了迭代器的使用寿命和重复可用性。状态放在实例属性里就能得到独立、可复现的迭代行为状态放在外部容器里迭代器就只是容器的“视图”会随容器变化而变化。根据你的业务需求选择合适的所有者。再想清楚“是否需要懒加载”。如果数据集很小一次性生成列表可能更简单直接如果数据集很大或是无限的懒加载几乎是唯一选择。生成器的惰性求值虽然节省内存但它也会带来“消费一次就没了”的约束。在API设计上最好用命名给调用方明确提示比如iter_*前缀表示返回迭代器*_list后缀表示返回完整列表。然后想清楚“异常和清理怎么处理”。无论是类实现的StopIteration还是生成器里的break、return都要保证遍历结束无论正常还是异常后资源能被正确释放。在生成器中优先用with语句管理文件、连接等外部资源能省掉大量手动清理的样板代码。最后是测试。自定义迭代器建议专门写测试覆盖以下几种情况迭代器被完整消费后的状态、提前中断后再次调用next()的行为、迭代过程中抛出异常时是否有资源泄漏、同一个迭代器能否被重复遍历等。这些边界情况是迭代器设计中最容易出问题、也最容易在代码评审中被忽略的部分。我在实际开发中还经常用“迭代器模式”来封装那些原本分散在业务代码里的遍历细节。比如从数据库分页查询数据、调用分页API拉取远端资源、读取多个日志文件做配对处理等。底层无论数据源多复杂对外暴露的都只是一个干净的迭代器调用方只需要for循环内部的翻页、重试、编码转换全部封装在迭代器里。这种封装方式能让业务代码的可读性上一个台阶。