我当年第一次看懂for循环内部的运行机制时有一种“原来如此”的顿悟感。Python 里的迭代器Iterator就像一条流水线而for循环只是按顺序从流水线上取零件的工人——它并不关心这条流水线背后是仓库里堆好的零件还是机器正在实时生产。Python 迭代器Iterator之所以值得花时间彻底搞懂是因为它直接影响你写出的代码是“一次性加载所有数据”还是“用多少取多少”。对于处理超大文件、无限序列、数据流这类场景理解迭代器几乎是必备技能。这篇内容适合两类人一类是刚学完 Python 基础语法、想更进一步理解语言底层设计的朋友另一类是已经写过不少 Python 代码、但遇到“生成器到底有什么用”“自定义可迭代对象怎么实现”时会卡壳的开发者。我会从for循环背后的协议机制讲起再手写一个迭代器最后聊一聊生成器、常用内置工具和一些容易踩的坑。1. 迭代器是什么一个“按需生产”的取值协议迭代器Iterator在 Python 中不是某个具体的类或函数而是一套约定、一套协议。用生活里的话说它就像一张只允许向前翻、不允许回看的“抽卡片”规则你每次只能拿下一张拿完就没有了不能倒回去重新拿。1.1 可迭代对象与迭代器的本质区别很多入门资料把“可迭代对象Iterable”和“迭代器Iterator”混着说这其实是个隐患。可迭代对象是能被for循环遍历的东西比如列表、元组、字典、字符串、集合。迭代器则是“能记住遍历位置”的对象它通过__next__()方法一个接一个地返回数据。有个比喻特别容易记可迭代对象是一摞书迭代器是这摞书上的一个书签指针。你可以把书签放到任意一本可迭代对象上通过它一页一页往后翻。关键在于书签是有状态的——它记得自己翻到哪了。这也是迭代器最核心的特征有状态的对象每次调用next()都会往前移动一步。在 Python 里可迭代对象和迭代器通过两个核心方法区分可迭代对象实现__iter__()该方法返回一个迭代器。迭代器实现__next__()该方法返回下一个元素没元素时抛出StopIteration。迭代器同时也要实现__iter__()返回自身这样迭代器本身也能用在for循环里。我工作里遇到过不少把“可迭代”和“迭代器”混为一谈导致的 bug。最典型的就是有人以为一个列表被遍历两次没问题结果发现第二次遍历什么都拿不到。如果你把列表传入iter()得到迭代器然后用for循环遍历完再跑一次for第二次循环直接就结束了——因为迭代器已经“走到头”了。而列表本身是支持反复遍历的因为它每次调用__iter__()都会生成一个全新的迭代器。1.2 为什么需要迭代器懒加载与内存优化如果你在处理一个 10GB 的日志文件直接readlines()会把整个文件加载进内存大概率直接卡死。用迭代器你可以逐行读取内存占用始终只保留当前这一行。这就是“懒加载lazy loading”的核心价值迭代器按需生成数据而不是一次性把数据全部准备好。初次接触这个概念时我把它理解成“自助餐厅的现切烤肉”和“打包好的盒饭”的区别。盒饭是一次性把所有菜都盛好列表现切烤肉是你吃一片、师傅切一片迭代器。处理小数据两者差不多但当数据量以 GB 计或者数据本身是无穷无尽的比如传感器持续产生的读数迭代器就是唯一可行的方案。除了内存优势迭代器还天然适合“流式处理”的编程模型。你做数据处理管道时从数据库游标里一条条取数据、处理完再丢出去第一想法就应该是迭代器而不是把全表数据先塞进列表再处理。这也是专业 Python 开发者和刚入门的人一个很直观的思维分水岭——你是先想到“把所有数据都拿到手上”还是先想到“用一条管线让数据流过”。2. for 循环背后的协议机制iter()与next()的协作for循环是 Python 中最常用的语法之一但你有没有想过它到底是怎么工作的简单说for循环做的事情就是先从可迭代对象拿到一个迭代器然后不断调用next()直到捕获StopIteration异常并结束循环。2.1 for 循环的底层展开一个 while 循环的等价实现直接看代码可能更清晰。你在 Python 里写for item in [1, 2, 3]: print(item)实际发生的事情等价于下面这段逻辑# for 循环的内部工作机制等价伪代码 it iter([1, 2, 3]) while True: try: item next(it) except StopIteration: # 迭代器耗尽跳出循环 break else: print(item)整个过程分为三步调用iter(可迭代对象)拿到迭代器。循环调用next(迭代器)获得下一个元素。当迭代器抛出StopIteration异常时循环捕获并正常结束。for循环之所以能遍历各种各样的数据类型正是因为它只依赖这两层协议目标对象能通过iter()返回迭代器迭代器能通过next()返回下一个元素。这就是著名的“鸭子类型”在遍历领域的体现——只要实现了这两个协议不管你是列表、文件、数据库游标还是自定义类统统可以用同一个for循环处理。2.2 常见的可迭代对象和迭代器实例Python 里最常见的可迭代对象包括序列类型list、tuple、str、range集合类型dict、set、frozenset文件对象open()返回的文件对象本身就可迭代逐行读取用的就是这个特性视图对象dict.keys()、dict.values()、dict.items()生成器由生成器函数或生成器表达式产生而典型的迭代器实例则包括iter([1, 2, 3])的返回值zip()、enumerate()、map()、filter()的返回值itertools模块中大多数函数返回的迭代器文件对象本身其实也是迭代器逐行迭代时用的是同一个对象这里要特别注意list不是迭代器。你可以这样验证my_list [1, 2, 3] # 列表本身没有 __next__ 方法 # next(my_list) # TypeError: list object is not an iterator # 但通过 iter() 可以得到迭代器 it iter(my_list) print(next(it)) # 1 print(next(it)) # 2 print(next(it)) # 3 # 再调用一次就会抛 StopIteration很多人刚开始会犯一个错误以为列表本身能直接调用next()。不行这必须通过iter()转换。而文件对象不一样它就是迭代器直接next(f)是完全合法的。2.3 for 循环在字节码层面是怎么跑的如果只想“用会”迭代器上面的内容已经够了。但如果你对“为什么”特别感兴趣可以简单看看for循环在 CPython 字节码层面的实现。用dis模块反汇编一下import dis def loop(): total 0 for i in range(10): total i return total dis.dis(loop)输出里会包含几个关键字节码指令GET_ITER负责调用iter()FOR_ITER则是循环的核心指令它每一步都会调用迭代器的__next__()方法当遇到StopIteration时跳转到循环结束的位置。看到GET_ITER和FOR_ITER这两个指令你就明白for循环的速度上限受限于迭代器__next__()的实现效率。这也是为什么纯 Python 自定义迭代器通常没有内置列表遍历快的原因——内置类型的__next__()是在 C 层面实现的而你的自定义迭代器要经过 Python 解释器一层。3. 动手实现迭代器从零写一个自定义迭代器类理论讲完来点实战。自己实现一个迭代器类并不难核心就是实现__iter__()和__next__()两个方法。3.1 用类实现一个“倒计时”迭代器假设我们要实现一个倒计时迭代器初始值从 5 开始每次next()减 1减到 0 结束class Countdown: 从 start 倒数到 0 的迭代器 def __init__(self, start): self.current start def __iter__(self): # 迭代器对象自身就是迭代器所以返回 self return self def __next__(self): if self.current 0: raise StopIteration value self.current self.current - 1 return value # 使用示例 countdown Countdown(3) for num in countdown: print(num) # 输出3, 2, 1, 0注意几个细节__next__()在数据耗尽时必须抛出StopIteration这是协议的一部分。不能在耗尽时返回None或者返回一个特殊值否则for循环不会感知到结束。__iter__()返回self即可因为这个类本身既是可迭代对象也是迭代器。如果你希望这个类能支持“每次 for 都从头开始”就需要在__iter__()里重置状态而不是简单返回self了。一旦迭代器被for循环遍历完它就到了末尾。如果你再次对同一个实例做for循环你会发现直接什么都不输出——因为迭代器已经耗尽了。这最后一个特性经常让人困扰。解决方式是在__iter__()里重置状态class RestartableCountdown: 支持重新开始遍历的计数器 def __init__(self, start): self.start start self.current start def __iter__(self): self.current self.start # 每次 for 循环重新开始 return self def __next__(self): if self.current 0: raise StopIteration value self.current self.current - 1 return value这样每次调用for都会从头开始因为for循环最开始会调用iter(obj)也就是__iter__()方法在方法里重置了计数器位置。3.2 一个更实用的例子按批次读取大文件实际工作中你很少只写一个倒计时器。更常见的场景是按批次处理大文件并希望把“取数据”的逻辑封装起来。比如你想从一个大 CSV 文件里按指定行数批量读取class CSVChunkReader: 按行批次读取 CSV 文件的自定义迭代器简化版 def __init__(self, file_path, chunk_size10): self.file_path file_path self.chunk_size chunk_size self.file None def __iter__(self): self.file open(self.file_path, r, encodingutf-8) return self def __next__(self): if self.file is None: raise RuntimeError(请先调用 iter() 初始化文件对象) lines [] for _ in range(self.chunk_size): line self.file.readline() if line: lines.append(line.strip()) else: break if not lines: self.file.close() raise StopIteration return lines # 使用示例 for chunk in CSVChunkReader(large_log.txt, chunk_size5): # 每次只处理 5 行避免一次性加载整个文件 process_chunk(chunk)这个例子展示了迭代器的一个精妙之处打开文件、关闭文件的时机可以完全交给迭代器协议来管理。for循环开始时调用__iter__()文件被打开循环正常结束时__next__()抛出StopIteration文件在此时被关闭。这个模式比我以前把所有逻辑写在大循环里要清晰很多也更好测试。3.3 用__getitem__实现迭代协议一种更简单的“偷懒”方式严格来说一个对象不实现__iter__()也能被for循环遍历只要它实现了__getitem__()方法并且按整数索引取值越界时抛出IndexError。Python 的迭代协议会退而求其次用__getitem__()逐个索引取值直到捕获IndexError。class FibonacciSequence: 用 __getitem__ 实现的斐波那契数列支持 for 循环遍历 def __init__(self, count): self.count count def __getitem__(self, index): if index self.count: raise IndexError(超出范围) if index in (0, 1): return index a, b 0, 1 for _ in range(index): a, b b, a b return a # for 循环会一直按索引 0, 1, 2... 调用 __getitem__ for num in FibonacciSequence(7): print(num) # 输出0, 1, 1, 2, 3, 5, 8这种方式的好处是代码更少不需要维护状态每次按索引计算值即可。缺点是每次按索引取值都要重新计算前面的值效率不高而且它和“有状态的迭代器”理念不同——这里更像是“随机访问容器”而非“流式生成”。不过对很多场景来说这种实现已经够用了而且可读性很好。4. 生成器写迭代器最优雅的姿势手写一个包含__iter__()和__next__()的类完全没问题但日常开发中你大概率不会这么做。因为 Python 提供了生成器Generator这个语法糖——任何包含yield关键字的函数都是生成器函数调用它不会立即执行函数体而是返回一个生成器对象它本身就是一个迭代器。4.1 生成器函数用 yield 代替 returndef countdown(start): 生成器版本的倒计时 while start 0: yield start start - 1 # 调用生成器函数不会立即执行而是返回生成器对象 gen countdown(3) print(gen) # generator object countdown at 0x... for num in gen: print(num) # 输出3, 2, 1, 0生成器函数和普通函数的区别在于执行流程的“暂停”与“恢复”。每次执行到yield时函数就把当前值返回给调用方并且记住当前的状态——所有局部变量、执行位置都会保存下来。下次调用next()时函数从上次yield的地方继续执行。打个比方普通函数是一口气跑完的百米冲刺生成器函数则是可以随时叫停、又可以随时接着跑的定向越野。每次叫停yield你会拿到一个有价值的坐标点下次接着跑时你仍然站在原地继续前进。4.2 生成器表达式列表推导式的懒加载兄弟生成器表达式和列表推导式长得很像只是把方括号换成圆括号# 列表推导式立即计算所有元素 squares_list [x * x for x in range(10)] # 生成器表达式惰性计算每次取一个 squares_gen (x * x for x in range(10)) print(sum(squares_gen)) # 285列表推导式会把整个列表创建出来占用内存生成器表达式则是一个迭代器每次按需计算。处理海量数据时用生成器表达式可以显著降低内存占用。我经常用生成器表达式配合sum()、max()、min()这类聚合函数因为它们只遍历一次配合惰性计算非常合适。不过要提醒一句如果数据量不大比如就几十个、几百个元素列表推导式的性能往往优于生成器表达式因为生成器有逐次调用的开销。只有数据量大或者不确定数据规模时才需要优先考虑生成器。4.3 yield 的高级用法send()、throw() 与 close()yield不仅能产出值还能接收值。通过在生成器上调用send(value)可以把值传回生成器内部这个值会成为当前yield表达式的返回值。常用来实现协程coroutine、状态机或数据流水线。def echo(): 协程示例接收输入并返回回声 while True: received yield # 接收 send() 传进来的值 yield f收到{received} gen echo() next(gen) # 启动生成器执行到第一个 yield print(gen.send(你好)) # 收到你好虽然现在asyncio是异步编程的标准方案但理解send()和yield的关系对你理解协程的底层原理大有帮助。此外生成器还有throw()方法从内部抛异常和close()方法关闭生成器不过日常使用频率不高遇到时知道有这么个方法就够了。4.4 迭代器与生成器的关系辨析迭代器是一个更宽泛的概念生成器是迭代器的一种具体实现。所有生成器都是迭代器但反过来不成立——你完全可以手写一个迭代器类而不使用yield。两者的区别可以这样概括对比项迭代器类实现生成器实现方式定义类实现__iter__()和__next__()定义包含yield的函数或使用生成器表达式代码复杂度较高需要维护状态、处理异常很低状态保存由解释器自动完成可读性一般逻辑分散在方法里很高执行流程线性直观适用场景需要完整控制状态、实现复杂协议时大多数普通场景过去几年我写代码的经验是90% 的情况下生成器就够了只有极少场景需要手写迭代器类——比如你要实现一个协议复杂的数据结构或者需要同时维护多组状态、控制重置行为时。5. 常用内置迭代器工具itertools 与内置函数Python 标准库里有一批非常趁手的迭代器工具用好了能让代码既简洁又高效。最值得掌握的是itertools模块和几个内置函数。5.1 正规军itertools 模块核心函数itertools是 Python 标准库中专门为迭代器设计的工具箱。我这里挑几个高频的讲from itertools import count, cycle, repeat, chain, islice # count(start, step)无限递增的计数器 for i in count(10, 2): if i 20: break print(i) # 输出10, 12, 14, 16, 18, 20 # cycle(iterable)无限循环迭代 colors cycle([red, green, blue]) for _ in range(6): print(next(colors)) # 输出red, green, blue, red, green, blue # repeat(element, times)重复产生同一元素 for item in repeat(A, 3): print(item) # 输出A, A, A # chain(iter1, iter2)把多个可迭代对象串起来 for item in chain([1, 2], [a, b]): print(item) # 输出1, 2, a, b # islice(iterable, start, stop)对迭代器切片但不生成列表 words iter([a, b, c, d, e]) for item in islice(words, 1, 4): print(item) # 输出b, c, d最让我惊喜的是islice。它相当于迭代器的切片工具但不需要把所有数据先转成列表。处理超大集合时islice能帮你只取中间某一段数据进行调试而不必整体载入内存。itertools里还有几个相当实用的高级函数比如groupby按 key 分组相邻元素、product笛卡尔积、permutations全排列、combinations组合。面试里偶尔会让你手写排列组合用itertools可以直接一行搞定。5.2 更易用的内置函数zip、enumerate、map、filter、range这几个内置函数返回的都是迭代器或类迭代器对象使用频率极高# zip并行迭代多个可迭代对象 names [Alice, Bob, Charlie] scores [88, 92, 85] for name, score in zip(names, scores): print(f{name}: {score}) # enumerate遍历时同时得到索引和值 for idx, name in enumerate(names, start1): print(idx, name) # map对每个元素应用函数返回迭代器 doubled map(lambda x: x * 2, [1, 2, 3]) print(list(doubled)) # [2, 4, 6] # filter按条件筛选元素返回迭代器 evens filter(lambda x: x % 2 0, range(10)) print(list(evens)) # [0, 2, 4, 6, 8] # range惰性生成整数序列Python 3 中不返回列表 nums range(1000000) print(len(nums)) # 1000000range 有自己的长度计算方式zip有个特性值得单独提它返回的迭代器耗尽后就会消失。如果你后面还需要这组配对数据记得存成列表或字典。我曾经在数据处理流程中踩过坑——zip的结果被一个for循环消费完了后面想再用同一个zip变量结果发现什么都取不到。5.3 迭代器与 iter(callable, sentinel) 的组合技巧iter()还有一个两种参数的重载形式iter(callable, sentinel)其中第一个参数是不断调用的可调用对象第二个参数是哨兵值——当调用结果等于哨兵值时停止。这个用法对持续读取直到某个标记出现的场景很实用# 从用户输入中持续读取直到遇到空行 lines [] for line in iter(input, ): # 空字符串作为哨兵值 lines.append(line) # 从二进制文件流中按固定大小读取块 with open(data.bin, rb) as f: for chunk in iter(lambda: f.read(4096), b): process_chunk(chunk)第二个例子很实用f.read(4096)每次最多读 4096 字节读到文件末尾返回空字节串b时循环自动终止。这个写法比我经常见到的while True break要简洁很多而且意图一目了然。6. 常见问题与排查技巧实录迭代器和生成器虽然不难但有些坑是每个 Python 开发者都会遇到的。这里整理几个我踩过的、也见过别人踩过的高频问题。6.1 迭代器被遍历一次后就“空了”这是最高频的问题。很多新手甚至熟练工会把迭代器当成列表以为可以反复遍历。事实是迭代器是一次性消耗品。第一次for循环已经把它“榨干”了第二次自然什么都取不到。numbers iter([1, 2, 3]) # 第一次遍历 for n in numbers: print(n) # 输出1, 2, 3 # 第二次遍历无输出 for n in numbers: print(n) # 因为迭代器已经耗尽需要重新创建一个 numbers iter([1, 2, 3])解决办法取决于你的需求。如果需要多次完整遍历直接用原始的可迭代对象比如列表而不是迭代器如果数据太大不能重复加载就重新创建迭代器。还有一个小技巧如果你不确定一个对象到底是迭代器还是可迭代对象可以用iter(obj) is obj来判断——如果是True说明obj就是迭代器它只能遍历一次。6.2 在遍历列表时修改列表导致的奇怪行为遍历列表的同时向列表添加或删除元素会引发各种诡异问题。比如nums [1, 2, 3, 4, 5] for n in nums: if n % 2 0: nums.remove(n) print(nums) # 实际输出会让你意外[1, 3, 4, 5]原因很简单迭代器通过索引追踪当前位置删除元素后后面的元素会“补位”导致迭代器跳过某些元素。解决这个问题最推荐的做法是迭代原始列表的副本或者使用列表推导式筛选nums [1, 2, 3, 4, 5] nums [n for n in nums if n % 2 ! 0] print(nums) # [1, 3, 5] # 或者反向遍历并删除 nums [1, 2, 3, 4, 5] for n in reversed(nums): if n % 2 0: nums.remove(n)用字典也有类似的坑遍历dict的同时修改它的键会直接抛RuntimeError: dictionary changed size during iteration。处理这类问题先想清楚“能不能用新的数据结构替换旧的”这通常是最干净、最不容易出错的方案。6.3 StopIteration 的误区不是所有“取不到”都要靠它自定义迭代器时StopIteration代表“正常耗尽”不是错误。但有几种情况会出现问题手写__next__()时有人会在耗尽时返回None而不是抛StopIteration导致for循环永远无法正确结束或者循环多跑几次。在生成器里return一个值这个值不会直接给你而是作为StopIteration异常的value属性存在。Python 3.7 及以后版本中你不应该主动去捕获这个StopIteration来拿返回值——这被官方认为是反模式因为会影响 PEP 479 的语义。如果你在生成器内部捕获了StopIteration却没有处理生成器可能直接停止而不是继续。所以最好只在明确合法的情况下处理这个异常。在 Python 3.7 中如果你在生成器内部隐式或显式地让StopIteration从生成器逃逸解释器会把它转成RuntimeError避免因迭代器协议和协程协议混用而产生难以察觉的 bug。这意味着你要优雅获取生成器的最终返回值应使用try/except StopIteration包裹显式的next()调用而不是在生成器体内乱捕获异常。6.4 生成器一直占用内存不释放生成器是懒加载的但如果你持有对生成器对象的引用它的栈帧包括所有局部变量就会一直留在内存里。这在长生命周期程序里会造成隐性内存增长。def read_lines(filepath): with open(filepath, r) as f: for line in f: yield line # 如果这个生成器一直被引用而不遍历文件也一直不会被关闭 gen read_lines(huge_file.log) # 使用完记得手动 close() gen.close()拿到生成器后如果决定不使用记得调用close()方法释放资源。用上下文管理器with语句配合contextlib.closing也能达到同样效果。这一点在小脚本里无关紧要但在需要长时间运行的常驻服务中不注意就会变成内存泄漏。6.5 区分“迭代器”和“生成器”时的判断技巧如果你想快速判断一个对象是不是迭代器使用isinstance(obj, collections.abc.Iterator)是最严谨的方式from collections.abc import Iterator, Iterable my_list [1, 2, 3] my_iter iter(my_list) gen (x for x in range(3)) print(isinstance(my_list, Iterable)) # True print(isinstance(my_list, Iterator)) # False print(isinstance(my_iter, Iterator)) # True print(isinstance(gen, Iterator)) # TrueIterable和Iterator是collections.abc里的抽象基类。Python 官方建议用它们做类型判断既清晰又符合协议设计原则。类型检查工具比如mypy也能理解这套抽象写出更准确的类型标注。7. 选型思考自定义迭代器、生成器还是迭代工具很多初学者会纠结同一件事可以用列表完成也可以用生成器完成还可以用itertools完成到底选哪个我的经验是分三层来思考可读性、性能、内存约束。7.1 三种方案的适用场景对比方案优点缺点典型场景列表 for 循环简单直接可重复遍历支持索引访问数据量大时内存占用高小数据集、需要多轮遍历、需要随机访问生成器惰性计算内存友好代码简洁只能遍历一次无法索引调试困难大文件、无限序列、数据流、管道处理itertools 工具专为迭代器设计操作灵活高效很多函数返回迭代器需要注意一次性消费组合/切片/分组、延迟计算等这个表格是我写代码时心里默认的一个决策参考。日常处理的数据不超过几千条时我会直接用列表因为它调试方便。数据量上到十万、百万级或者不知道上限时改用生成器。需要复杂迭代操作时再到itertools里找有没有现成函数避免重复造轮子。7.2 一个面试常问的内存对比实测以前我自己对“生成器省内存”只有模糊的概念直到亲手做了个实验才真正有体会。用sys.getsizeof对比一下同样数列在列表和生成器中的内存占用import sys list_squares [x * x for x in range(1000)] gen_squares (x * x for x in range(1000)) print(sys.getsizeof(list_squares)) # 列表本身占用的字节数 print(sys.getsizeof(gen_squares)) # 生成器对象占用的字节数在我的机器上1000 个平方数的列表大约占 8856 字节而生成器对象本身只有 112 字节左右。当然生成器每次生成元素也要时间这是“用 CPU 换内存”的权衡。如果数据放在内存里完全没问题用生成器反而可能更慢数据大到内存装不下生成器就是唯一选择。7.3 调试迭代器与生成器的实用技巧调试生成器比调试普通函数难因为你不能直接看到“当前执行到哪了”。我的经验是转成列表再调试小规模数据时用list(generator)把生成器转成列表能直观看到全部产出值。在 yield 前打印关键变量这在排查生成器内部状态时非常管用。用tee克隆迭代器做多路消费itertools.tee()可以把一个迭代器复制成 n 个独立的迭代器便于同时做不同的处理。from itertools import tee numbers iter([1, 2, 3, 4, 5]) first, second tee(numbers) # 克隆出两个迭代器 print(list(first)) # [1, 2, 3, 4, 5] print(list(second)) # [1, 2, 3, 4, 5]tee()内部会缓存数据所以不能节省内存但在“同一份数据流需要做多路分析”的场景下非常实用。比如日志流可以复制一份给告警模块一份给统计模块。8. 进阶实践用迭代器改写一个数据处理管道分享一个我实际做过的数据管道重构案例。最开始拿到一批日志文件每个文件有几百万行需要按行解析、过滤、提取关键字段最后按时间窗口聚合成统计结果。最初的实现是先用readlines()读取全部行再用列表推导式做各种处理。结果程序跑到一半内存占用飙升服务器直接卡死。后来我重构为基于生成器的管道模型import re from itertools import groupby def read_log_lines(filepath): 逐行读取日志文件生成器 with open(filepath, r, encodingutf-8) as f: for line in f: yield line def parse_line(line): 解析单行日志提取时间戳和日志级别格式不合法则跳过 pattern r\[(?Ptime.*?)\] \[(?Plevel\w)\] (?Pmessage.*) match re.match(pattern, line) if match: return (match.group(time)[:13], match.group(level), match.group(message)) return None def filter_error(parsed): 只保留 ERROR 级别的日志 return parsed is not None and parsed[1] ERROR # 组装管道 log_path application.log parsed filter(None, map(parse_line, read_log_lines(log_path))) errors filter(lambda x: x[1] ERROR, parsed) grouped groupby(errors, keylambda x: x[0]) # 按小时分组 # 逐组处理 for hour, items in grouped: error_list list(items) print(f小时 {hour}错误数{len(error_list)})重构后内存占用从几个 GB 降到了几十 MB整个管道是流式的read_log_lines读一行map解析一行filter过滤一行groupby累积同一小时的计数——没有任何一个环节会一次性加载全部数据。这也算是迭代器在实际工程里最大的价值体现。管道的设计思路很像流水线车间每个函数只负责一个工序产物直接送给下一个工序。中间不需要把半成品搬进仓库大大减少了搬运成本。这种编程方式在数据处理、日志分析、网络包处理等场景中几乎是标配。9. 迭代器在框架和底层设计中的身影如果你写过 Django 的分页查询、用过pandas.read_csv()读取超大 CSV、或者用过requests库的流式下载你其实已经在使用迭代器的思想了。但迭代器的作用范围远不止这些。9.1 ORM 和数据库查询中的“游标”很多 Web 框架的数据库 ORM对象关系映射都支持惰性查询。以 Django ORM 为例# Django ORM 的惰性查询 all_users User.objects.all() # 此时并没有执行 SQL for user in all_users.iterator(): # 用 iterator() 强制流式取数据 process(user).iterator()方法会使用数据库游标逐条取数据而不是一次把整个结果集加载进内存。这在处理百万级用户数据做数据迁移或批量导出时特别重要。如果不用迭代器查询结果会把所有行先加载到内存分分钟打爆 RDS 实例的内存。9.2 流式协议与“无限序列”网络编程里经常用到无限序列的概念。比如实现一个心跳包生成器不需要存一个巨大的列表来代表“无限心跳”只需要一个永不耗尽的生成器def heartbeat(interval1): seq 0 while True: seq 1 yield {seq: seq, timestamp: time.time()} time.sleep(interval) for packet in heartbeat(): send(packet)这种模式在物联网数据采集、流媒体处理、行情推送等场景中大量出现。迭代器天然支持“不知道什么时候结束的数据流”让代码不需要为“边界条件”预设上限。9.3 扩展了解 asyncio 与异步迭代器如果你已经掌握了同步迭代器可以进一步了解异步迭代器Async Iterator。Python 3.6 引入了__aiter__()和__anext__()协议配合async for使用。比如用aiohttp异步读取网页内容import aiohttp import asyncio async def fetch_pages(urls): async with aiohttp.ClientSession() as session: for url in urls: async with session.get(url) as resp: yield await resp.text() async def main(): async for content in fetch_pages([https://example.com, https://example.org]): print(len(content)) asyncio.run(main())异步迭代器的思路和同步迭代器完全一致只是取值过程不阻塞线程。理解了同步迭代器和for循环背后的机制再看async for会非常轻松——你只是把原来同步的next()换成了可等待的__anext__()而已。迭代器的应用远不止“省内存”这么简单——它是一种组合数据流的抽象方式是构建可扩展程序的基石。把for循环真正搞明白之后你对 Python 的理解会上升一个台阶写出来的代码也会更接近 Python 的设计哲学简洁、清晰、优雅。如果这个内容对你有所帮助强烈建议你打开终端把上面的代码亲手跑一遍。先写一个最简单的生成器再尝试写一个自定义迭代器类最后用itertools组合出更复杂的数据管道。只有自己写一遍你才能真正理解迭代器为什么是 Python 中最优雅的设计之一。 深入理解Python迭代器与生成器：从for循环到底层协议
我当年第一次看懂for循环内部的运行机制时有一种“原来如此”的顿悟感。Python 里的迭代器Iterator就像一条流水线而for循环只是按顺序从流水线上取零件的工人——它并不关心这条流水线背后是仓库里堆好的零件还是机器正在实时生产。Python 迭代器Iterator之所以值得花时间彻底搞懂是因为它直接影响你写出的代码是“一次性加载所有数据”还是“用多少取多少”。对于处理超大文件、无限序列、数据流这类场景理解迭代器几乎是必备技能。这篇内容适合两类人一类是刚学完 Python 基础语法、想更进一步理解语言底层设计的朋友另一类是已经写过不少 Python 代码、但遇到“生成器到底有什么用”“自定义可迭代对象怎么实现”时会卡壳的开发者。我会从for循环背后的协议机制讲起再手写一个迭代器最后聊一聊生成器、常用内置工具和一些容易踩的坑。1. 迭代器是什么一个“按需生产”的取值协议迭代器Iterator在 Python 中不是某个具体的类或函数而是一套约定、一套协议。用生活里的话说它就像一张只允许向前翻、不允许回看的“抽卡片”规则你每次只能拿下一张拿完就没有了不能倒回去重新拿。1.1 可迭代对象与迭代器的本质区别很多入门资料把“可迭代对象Iterable”和“迭代器Iterator”混着说这其实是个隐患。可迭代对象是能被for循环遍历的东西比如列表、元组、字典、字符串、集合。迭代器则是“能记住遍历位置”的对象它通过__next__()方法一个接一个地返回数据。有个比喻特别容易记可迭代对象是一摞书迭代器是这摞书上的一个书签指针。你可以把书签放到任意一本可迭代对象上通过它一页一页往后翻。关键在于书签是有状态的——它记得自己翻到哪了。这也是迭代器最核心的特征有状态的对象每次调用next()都会往前移动一步。在 Python 里可迭代对象和迭代器通过两个核心方法区分可迭代对象实现__iter__()该方法返回一个迭代器。迭代器实现__next__()该方法返回下一个元素没元素时抛出StopIteration。迭代器同时也要实现__iter__()返回自身这样迭代器本身也能用在for循环里。我工作里遇到过不少把“可迭代”和“迭代器”混为一谈导致的 bug。最典型的就是有人以为一个列表被遍历两次没问题结果发现第二次遍历什么都拿不到。如果你把列表传入iter()得到迭代器然后用for循环遍历完再跑一次for第二次循环直接就结束了——因为迭代器已经“走到头”了。而列表本身是支持反复遍历的因为它每次调用__iter__()都会生成一个全新的迭代器。1.2 为什么需要迭代器懒加载与内存优化如果你在处理一个 10GB 的日志文件直接readlines()会把整个文件加载进内存大概率直接卡死。用迭代器你可以逐行读取内存占用始终只保留当前这一行。这就是“懒加载lazy loading”的核心价值迭代器按需生成数据而不是一次性把数据全部准备好。初次接触这个概念时我把它理解成“自助餐厅的现切烤肉”和“打包好的盒饭”的区别。盒饭是一次性把所有菜都盛好列表现切烤肉是你吃一片、师傅切一片迭代器。处理小数据两者差不多但当数据量以 GB 计或者数据本身是无穷无尽的比如传感器持续产生的读数迭代器就是唯一可行的方案。除了内存优势迭代器还天然适合“流式处理”的编程模型。你做数据处理管道时从数据库游标里一条条取数据、处理完再丢出去第一想法就应该是迭代器而不是把全表数据先塞进列表再处理。这也是专业 Python 开发者和刚入门的人一个很直观的思维分水岭——你是先想到“把所有数据都拿到手上”还是先想到“用一条管线让数据流过”。2. for 循环背后的协议机制iter()与next()的协作for循环是 Python 中最常用的语法之一但你有没有想过它到底是怎么工作的简单说for循环做的事情就是先从可迭代对象拿到一个迭代器然后不断调用next()直到捕获StopIteration异常并结束循环。2.1 for 循环的底层展开一个 while 循环的等价实现直接看代码可能更清晰。你在 Python 里写for item in [1, 2, 3]: print(item)实际发生的事情等价于下面这段逻辑# for 循环的内部工作机制等价伪代码 it iter([1, 2, 3]) while True: try: item next(it) except StopIteration: # 迭代器耗尽跳出循环 break else: print(item)整个过程分为三步调用iter(可迭代对象)拿到迭代器。循环调用next(迭代器)获得下一个元素。当迭代器抛出StopIteration异常时循环捕获并正常结束。for循环之所以能遍历各种各样的数据类型正是因为它只依赖这两层协议目标对象能通过iter()返回迭代器迭代器能通过next()返回下一个元素。这就是著名的“鸭子类型”在遍历领域的体现——只要实现了这两个协议不管你是列表、文件、数据库游标还是自定义类统统可以用同一个for循环处理。2.2 常见的可迭代对象和迭代器实例Python 里最常见的可迭代对象包括序列类型list、tuple、str、range集合类型dict、set、frozenset文件对象open()返回的文件对象本身就可迭代逐行读取用的就是这个特性视图对象dict.keys()、dict.values()、dict.items()生成器由生成器函数或生成器表达式产生而典型的迭代器实例则包括iter([1, 2, 3])的返回值zip()、enumerate()、map()、filter()的返回值itertools模块中大多数函数返回的迭代器文件对象本身其实也是迭代器逐行迭代时用的是同一个对象这里要特别注意list不是迭代器。你可以这样验证my_list [1, 2, 3] # 列表本身没有 __next__ 方法 # next(my_list) # TypeError: list object is not an iterator # 但通过 iter() 可以得到迭代器 it iter(my_list) print(next(it)) # 1 print(next(it)) # 2 print(next(it)) # 3 # 再调用一次就会抛 StopIteration很多人刚开始会犯一个错误以为列表本身能直接调用next()。不行这必须通过iter()转换。而文件对象不一样它就是迭代器直接next(f)是完全合法的。2.3 for 循环在字节码层面是怎么跑的如果只想“用会”迭代器上面的内容已经够了。但如果你对“为什么”特别感兴趣可以简单看看for循环在 CPython 字节码层面的实现。用dis模块反汇编一下import dis def loop(): total 0 for i in range(10): total i return total dis.dis(loop)输出里会包含几个关键字节码指令GET_ITER负责调用iter()FOR_ITER则是循环的核心指令它每一步都会调用迭代器的__next__()方法当遇到StopIteration时跳转到循环结束的位置。看到GET_ITER和FOR_ITER这两个指令你就明白for循环的速度上限受限于迭代器__next__()的实现效率。这也是为什么纯 Python 自定义迭代器通常没有内置列表遍历快的原因——内置类型的__next__()是在 C 层面实现的而你的自定义迭代器要经过 Python 解释器一层。3. 动手实现迭代器从零写一个自定义迭代器类理论讲完来点实战。自己实现一个迭代器类并不难核心就是实现__iter__()和__next__()两个方法。3.1 用类实现一个“倒计时”迭代器假设我们要实现一个倒计时迭代器初始值从 5 开始每次next()减 1减到 0 结束class Countdown: 从 start 倒数到 0 的迭代器 def __init__(self, start): self.current start def __iter__(self): # 迭代器对象自身就是迭代器所以返回 self return self def __next__(self): if self.current 0: raise StopIteration value self.current self.current - 1 return value # 使用示例 countdown Countdown(3) for num in countdown: print(num) # 输出3, 2, 1, 0注意几个细节__next__()在数据耗尽时必须抛出StopIteration这是协议的一部分。不能在耗尽时返回None或者返回一个特殊值否则for循环不会感知到结束。__iter__()返回self即可因为这个类本身既是可迭代对象也是迭代器。如果你希望这个类能支持“每次 for 都从头开始”就需要在__iter__()里重置状态而不是简单返回self了。一旦迭代器被for循环遍历完它就到了末尾。如果你再次对同一个实例做for循环你会发现直接什么都不输出——因为迭代器已经耗尽了。这最后一个特性经常让人困扰。解决方式是在__iter__()里重置状态class RestartableCountdown: 支持重新开始遍历的计数器 def __init__(self, start): self.start start self.current start def __iter__(self): self.current self.start # 每次 for 循环重新开始 return self def __next__(self): if self.current 0: raise StopIteration value self.current self.current - 1 return value这样每次调用for都会从头开始因为for循环最开始会调用iter(obj)也就是__iter__()方法在方法里重置了计数器位置。3.2 一个更实用的例子按批次读取大文件实际工作中你很少只写一个倒计时器。更常见的场景是按批次处理大文件并希望把“取数据”的逻辑封装起来。比如你想从一个大 CSV 文件里按指定行数批量读取class CSVChunkReader: 按行批次读取 CSV 文件的自定义迭代器简化版 def __init__(self, file_path, chunk_size10): self.file_path file_path self.chunk_size chunk_size self.file None def __iter__(self): self.file open(self.file_path, r, encodingutf-8) return self def __next__(self): if self.file is None: raise RuntimeError(请先调用 iter() 初始化文件对象) lines [] for _ in range(self.chunk_size): line self.file.readline() if line: lines.append(line.strip()) else: break if not lines: self.file.close() raise StopIteration return lines # 使用示例 for chunk in CSVChunkReader(large_log.txt, chunk_size5): # 每次只处理 5 行避免一次性加载整个文件 process_chunk(chunk)这个例子展示了迭代器的一个精妙之处打开文件、关闭文件的时机可以完全交给迭代器协议来管理。for循环开始时调用__iter__()文件被打开循环正常结束时__next__()抛出StopIteration文件在此时被关闭。这个模式比我以前把所有逻辑写在大循环里要清晰很多也更好测试。3.3 用__getitem__实现迭代协议一种更简单的“偷懒”方式严格来说一个对象不实现__iter__()也能被for循环遍历只要它实现了__getitem__()方法并且按整数索引取值越界时抛出IndexError。Python 的迭代协议会退而求其次用__getitem__()逐个索引取值直到捕获IndexError。class FibonacciSequence: 用 __getitem__ 实现的斐波那契数列支持 for 循环遍历 def __init__(self, count): self.count count def __getitem__(self, index): if index self.count: raise IndexError(超出范围) if index in (0, 1): return index a, b 0, 1 for _ in range(index): a, b b, a b return a # for 循环会一直按索引 0, 1, 2... 调用 __getitem__ for num in FibonacciSequence(7): print(num) # 输出0, 1, 1, 2, 3, 5, 8这种方式的好处是代码更少不需要维护状态每次按索引计算值即可。缺点是每次按索引取值都要重新计算前面的值效率不高而且它和“有状态的迭代器”理念不同——这里更像是“随机访问容器”而非“流式生成”。不过对很多场景来说这种实现已经够用了而且可读性很好。4. 生成器写迭代器最优雅的姿势手写一个包含__iter__()和__next__()的类完全没问题但日常开发中你大概率不会这么做。因为 Python 提供了生成器Generator这个语法糖——任何包含yield关键字的函数都是生成器函数调用它不会立即执行函数体而是返回一个生成器对象它本身就是一个迭代器。4.1 生成器函数用 yield 代替 returndef countdown(start): 生成器版本的倒计时 while start 0: yield start start - 1 # 调用生成器函数不会立即执行而是返回生成器对象 gen countdown(3) print(gen) # generator object countdown at 0x... for num in gen: print(num) # 输出3, 2, 1, 0生成器函数和普通函数的区别在于执行流程的“暂停”与“恢复”。每次执行到yield时函数就把当前值返回给调用方并且记住当前的状态——所有局部变量、执行位置都会保存下来。下次调用next()时函数从上次yield的地方继续执行。打个比方普通函数是一口气跑完的百米冲刺生成器函数则是可以随时叫停、又可以随时接着跑的定向越野。每次叫停yield你会拿到一个有价值的坐标点下次接着跑时你仍然站在原地继续前进。4.2 生成器表达式列表推导式的懒加载兄弟生成器表达式和列表推导式长得很像只是把方括号换成圆括号# 列表推导式立即计算所有元素 squares_list [x * x for x in range(10)] # 生成器表达式惰性计算每次取一个 squares_gen (x * x for x in range(10)) print(sum(squares_gen)) # 285列表推导式会把整个列表创建出来占用内存生成器表达式则是一个迭代器每次按需计算。处理海量数据时用生成器表达式可以显著降低内存占用。我经常用生成器表达式配合sum()、max()、min()这类聚合函数因为它们只遍历一次配合惰性计算非常合适。不过要提醒一句如果数据量不大比如就几十个、几百个元素列表推导式的性能往往优于生成器表达式因为生成器有逐次调用的开销。只有数据量大或者不确定数据规模时才需要优先考虑生成器。4.3 yield 的高级用法send()、throw() 与 close()yield不仅能产出值还能接收值。通过在生成器上调用send(value)可以把值传回生成器内部这个值会成为当前yield表达式的返回值。常用来实现协程coroutine、状态机或数据流水线。def echo(): 协程示例接收输入并返回回声 while True: received yield # 接收 send() 传进来的值 yield f收到{received} gen echo() next(gen) # 启动生成器执行到第一个 yield print(gen.send(你好)) # 收到你好虽然现在asyncio是异步编程的标准方案但理解send()和yield的关系对你理解协程的底层原理大有帮助。此外生成器还有throw()方法从内部抛异常和close()方法关闭生成器不过日常使用频率不高遇到时知道有这么个方法就够了。4.4 迭代器与生成器的关系辨析迭代器是一个更宽泛的概念生成器是迭代器的一种具体实现。所有生成器都是迭代器但反过来不成立——你完全可以手写一个迭代器类而不使用yield。两者的区别可以这样概括对比项迭代器类实现生成器实现方式定义类实现__iter__()和__next__()定义包含yield的函数或使用生成器表达式代码复杂度较高需要维护状态、处理异常很低状态保存由解释器自动完成可读性一般逻辑分散在方法里很高执行流程线性直观适用场景需要完整控制状态、实现复杂协议时大多数普通场景过去几年我写代码的经验是90% 的情况下生成器就够了只有极少场景需要手写迭代器类——比如你要实现一个协议复杂的数据结构或者需要同时维护多组状态、控制重置行为时。5. 常用内置迭代器工具itertools 与内置函数Python 标准库里有一批非常趁手的迭代器工具用好了能让代码既简洁又高效。最值得掌握的是itertools模块和几个内置函数。5.1 正规军itertools 模块核心函数itertools是 Python 标准库中专门为迭代器设计的工具箱。我这里挑几个高频的讲from itertools import count, cycle, repeat, chain, islice # count(start, step)无限递增的计数器 for i in count(10, 2): if i 20: break print(i) # 输出10, 12, 14, 16, 18, 20 # cycle(iterable)无限循环迭代 colors cycle([red, green, blue]) for _ in range(6): print(next(colors)) # 输出red, green, blue, red, green, blue # repeat(element, times)重复产生同一元素 for item in repeat(A, 3): print(item) # 输出A, A, A # chain(iter1, iter2)把多个可迭代对象串起来 for item in chain([1, 2], [a, b]): print(item) # 输出1, 2, a, b # islice(iterable, start, stop)对迭代器切片但不生成列表 words iter([a, b, c, d, e]) for item in islice(words, 1, 4): print(item) # 输出b, c, d最让我惊喜的是islice。它相当于迭代器的切片工具但不需要把所有数据先转成列表。处理超大集合时islice能帮你只取中间某一段数据进行调试而不必整体载入内存。itertools里还有几个相当实用的高级函数比如groupby按 key 分组相邻元素、product笛卡尔积、permutations全排列、combinations组合。面试里偶尔会让你手写排列组合用itertools可以直接一行搞定。5.2 更易用的内置函数zip、enumerate、map、filter、range这几个内置函数返回的都是迭代器或类迭代器对象使用频率极高# zip并行迭代多个可迭代对象 names [Alice, Bob, Charlie] scores [88, 92, 85] for name, score in zip(names, scores): print(f{name}: {score}) # enumerate遍历时同时得到索引和值 for idx, name in enumerate(names, start1): print(idx, name) # map对每个元素应用函数返回迭代器 doubled map(lambda x: x * 2, [1, 2, 3]) print(list(doubled)) # [2, 4, 6] # filter按条件筛选元素返回迭代器 evens filter(lambda x: x % 2 0, range(10)) print(list(evens)) # [0, 2, 4, 6, 8] # range惰性生成整数序列Python 3 中不返回列表 nums range(1000000) print(len(nums)) # 1000000range 有自己的长度计算方式zip有个特性值得单独提它返回的迭代器耗尽后就会消失。如果你后面还需要这组配对数据记得存成列表或字典。我曾经在数据处理流程中踩过坑——zip的结果被一个for循环消费完了后面想再用同一个zip变量结果发现什么都取不到。5.3 迭代器与 iter(callable, sentinel) 的组合技巧iter()还有一个两种参数的重载形式iter(callable, sentinel)其中第一个参数是不断调用的可调用对象第二个参数是哨兵值——当调用结果等于哨兵值时停止。这个用法对持续读取直到某个标记出现的场景很实用# 从用户输入中持续读取直到遇到空行 lines [] for line in iter(input, ): # 空字符串作为哨兵值 lines.append(line) # 从二进制文件流中按固定大小读取块 with open(data.bin, rb) as f: for chunk in iter(lambda: f.read(4096), b): process_chunk(chunk)第二个例子很实用f.read(4096)每次最多读 4096 字节读到文件末尾返回空字节串b时循环自动终止。这个写法比我经常见到的while True break要简洁很多而且意图一目了然。6. 常见问题与排查技巧实录迭代器和生成器虽然不难但有些坑是每个 Python 开发者都会遇到的。这里整理几个我踩过的、也见过别人踩过的高频问题。6.1 迭代器被遍历一次后就“空了”这是最高频的问题。很多新手甚至熟练工会把迭代器当成列表以为可以反复遍历。事实是迭代器是一次性消耗品。第一次for循环已经把它“榨干”了第二次自然什么都取不到。numbers iter([1, 2, 3]) # 第一次遍历 for n in numbers: print(n) # 输出1, 2, 3 # 第二次遍历无输出 for n in numbers: print(n) # 因为迭代器已经耗尽需要重新创建一个 numbers iter([1, 2, 3])解决办法取决于你的需求。如果需要多次完整遍历直接用原始的可迭代对象比如列表而不是迭代器如果数据太大不能重复加载就重新创建迭代器。还有一个小技巧如果你不确定一个对象到底是迭代器还是可迭代对象可以用iter(obj) is obj来判断——如果是True说明obj就是迭代器它只能遍历一次。6.2 在遍历列表时修改列表导致的奇怪行为遍历列表的同时向列表添加或删除元素会引发各种诡异问题。比如nums [1, 2, 3, 4, 5] for n in nums: if n % 2 0: nums.remove(n) print(nums) # 实际输出会让你意外[1, 3, 4, 5]原因很简单迭代器通过索引追踪当前位置删除元素后后面的元素会“补位”导致迭代器跳过某些元素。解决这个问题最推荐的做法是迭代原始列表的副本或者使用列表推导式筛选nums [1, 2, 3, 4, 5] nums [n for n in nums if n % 2 ! 0] print(nums) # [1, 3, 5] # 或者反向遍历并删除 nums [1, 2, 3, 4, 5] for n in reversed(nums): if n % 2 0: nums.remove(n)用字典也有类似的坑遍历dict的同时修改它的键会直接抛RuntimeError: dictionary changed size during iteration。处理这类问题先想清楚“能不能用新的数据结构替换旧的”这通常是最干净、最不容易出错的方案。6.3 StopIteration 的误区不是所有“取不到”都要靠它自定义迭代器时StopIteration代表“正常耗尽”不是错误。但有几种情况会出现问题手写__next__()时有人会在耗尽时返回None而不是抛StopIteration导致for循环永远无法正确结束或者循环多跑几次。在生成器里return一个值这个值不会直接给你而是作为StopIteration异常的value属性存在。Python 3.7 及以后版本中你不应该主动去捕获这个StopIteration来拿返回值——这被官方认为是反模式因为会影响 PEP 479 的语义。如果你在生成器内部捕获了StopIteration却没有处理生成器可能直接停止而不是继续。所以最好只在明确合法的情况下处理这个异常。在 Python 3.7 中如果你在生成器内部隐式或显式地让StopIteration从生成器逃逸解释器会把它转成RuntimeError避免因迭代器协议和协程协议混用而产生难以察觉的 bug。这意味着你要优雅获取生成器的最终返回值应使用try/except StopIteration包裹显式的next()调用而不是在生成器体内乱捕获异常。6.4 生成器一直占用内存不释放生成器是懒加载的但如果你持有对生成器对象的引用它的栈帧包括所有局部变量就会一直留在内存里。这在长生命周期程序里会造成隐性内存增长。def read_lines(filepath): with open(filepath, r) as f: for line in f: yield line # 如果这个生成器一直被引用而不遍历文件也一直不会被关闭 gen read_lines(huge_file.log) # 使用完记得手动 close() gen.close()拿到生成器后如果决定不使用记得调用close()方法释放资源。用上下文管理器with语句配合contextlib.closing也能达到同样效果。这一点在小脚本里无关紧要但在需要长时间运行的常驻服务中不注意就会变成内存泄漏。6.5 区分“迭代器”和“生成器”时的判断技巧如果你想快速判断一个对象是不是迭代器使用isinstance(obj, collections.abc.Iterator)是最严谨的方式from collections.abc import Iterator, Iterable my_list [1, 2, 3] my_iter iter(my_list) gen (x for x in range(3)) print(isinstance(my_list, Iterable)) # True print(isinstance(my_list, Iterator)) # False print(isinstance(my_iter, Iterator)) # True print(isinstance(gen, Iterator)) # TrueIterable和Iterator是collections.abc里的抽象基类。Python 官方建议用它们做类型判断既清晰又符合协议设计原则。类型检查工具比如mypy也能理解这套抽象写出更准确的类型标注。7. 选型思考自定义迭代器、生成器还是迭代工具很多初学者会纠结同一件事可以用列表完成也可以用生成器完成还可以用itertools完成到底选哪个我的经验是分三层来思考可读性、性能、内存约束。7.1 三种方案的适用场景对比方案优点缺点典型场景列表 for 循环简单直接可重复遍历支持索引访问数据量大时内存占用高小数据集、需要多轮遍历、需要随机访问生成器惰性计算内存友好代码简洁只能遍历一次无法索引调试困难大文件、无限序列、数据流、管道处理itertools 工具专为迭代器设计操作灵活高效很多函数返回迭代器需要注意一次性消费组合/切片/分组、延迟计算等这个表格是我写代码时心里默认的一个决策参考。日常处理的数据不超过几千条时我会直接用列表因为它调试方便。数据量上到十万、百万级或者不知道上限时改用生成器。需要复杂迭代操作时再到itertools里找有没有现成函数避免重复造轮子。7.2 一个面试常问的内存对比实测以前我自己对“生成器省内存”只有模糊的概念直到亲手做了个实验才真正有体会。用sys.getsizeof对比一下同样数列在列表和生成器中的内存占用import sys list_squares [x * x for x in range(1000)] gen_squares (x * x for x in range(1000)) print(sys.getsizeof(list_squares)) # 列表本身占用的字节数 print(sys.getsizeof(gen_squares)) # 生成器对象占用的字节数在我的机器上1000 个平方数的列表大约占 8856 字节而生成器对象本身只有 112 字节左右。当然生成器每次生成元素也要时间这是“用 CPU 换内存”的权衡。如果数据放在内存里完全没问题用生成器反而可能更慢数据大到内存装不下生成器就是唯一选择。7.3 调试迭代器与生成器的实用技巧调试生成器比调试普通函数难因为你不能直接看到“当前执行到哪了”。我的经验是转成列表再调试小规模数据时用list(generator)把生成器转成列表能直观看到全部产出值。在 yield 前打印关键变量这在排查生成器内部状态时非常管用。用tee克隆迭代器做多路消费itertools.tee()可以把一个迭代器复制成 n 个独立的迭代器便于同时做不同的处理。from itertools import tee numbers iter([1, 2, 3, 4, 5]) first, second tee(numbers) # 克隆出两个迭代器 print(list(first)) # [1, 2, 3, 4, 5] print(list(second)) # [1, 2, 3, 4, 5]tee()内部会缓存数据所以不能节省内存但在“同一份数据流需要做多路分析”的场景下非常实用。比如日志流可以复制一份给告警模块一份给统计模块。8. 进阶实践用迭代器改写一个数据处理管道分享一个我实际做过的数据管道重构案例。最开始拿到一批日志文件每个文件有几百万行需要按行解析、过滤、提取关键字段最后按时间窗口聚合成统计结果。最初的实现是先用readlines()读取全部行再用列表推导式做各种处理。结果程序跑到一半内存占用飙升服务器直接卡死。后来我重构为基于生成器的管道模型import re from itertools import groupby def read_log_lines(filepath): 逐行读取日志文件生成器 with open(filepath, r, encodingutf-8) as f: for line in f: yield line def parse_line(line): 解析单行日志提取时间戳和日志级别格式不合法则跳过 pattern r\[(?Ptime.*?)\] \[(?Plevel\w)\] (?Pmessage.*) match re.match(pattern, line) if match: return (match.group(time)[:13], match.group(level), match.group(message)) return None def filter_error(parsed): 只保留 ERROR 级别的日志 return parsed is not None and parsed[1] ERROR # 组装管道 log_path application.log parsed filter(None, map(parse_line, read_log_lines(log_path))) errors filter(lambda x: x[1] ERROR, parsed) grouped groupby(errors, keylambda x: x[0]) # 按小时分组 # 逐组处理 for hour, items in grouped: error_list list(items) print(f小时 {hour}错误数{len(error_list)})重构后内存占用从几个 GB 降到了几十 MB整个管道是流式的read_log_lines读一行map解析一行filter过滤一行groupby累积同一小时的计数——没有任何一个环节会一次性加载全部数据。这也算是迭代器在实际工程里最大的价值体现。管道的设计思路很像流水线车间每个函数只负责一个工序产物直接送给下一个工序。中间不需要把半成品搬进仓库大大减少了搬运成本。这种编程方式在数据处理、日志分析、网络包处理等场景中几乎是标配。9. 迭代器在框架和底层设计中的身影如果你写过 Django 的分页查询、用过pandas.read_csv()读取超大 CSV、或者用过requests库的流式下载你其实已经在使用迭代器的思想了。但迭代器的作用范围远不止这些。9.1 ORM 和数据库查询中的“游标”很多 Web 框架的数据库 ORM对象关系映射都支持惰性查询。以 Django ORM 为例# Django ORM 的惰性查询 all_users User.objects.all() # 此时并没有执行 SQL for user in all_users.iterator(): # 用 iterator() 强制流式取数据 process(user).iterator()方法会使用数据库游标逐条取数据而不是一次把整个结果集加载进内存。这在处理百万级用户数据做数据迁移或批量导出时特别重要。如果不用迭代器查询结果会把所有行先加载到内存分分钟打爆 RDS 实例的内存。9.2 流式协议与“无限序列”网络编程里经常用到无限序列的概念。比如实现一个心跳包生成器不需要存一个巨大的列表来代表“无限心跳”只需要一个永不耗尽的生成器def heartbeat(interval1): seq 0 while True: seq 1 yield {seq: seq, timestamp: time.time()} time.sleep(interval) for packet in heartbeat(): send(packet)这种模式在物联网数据采集、流媒体处理、行情推送等场景中大量出现。迭代器天然支持“不知道什么时候结束的数据流”让代码不需要为“边界条件”预设上限。9.3 扩展了解 asyncio 与异步迭代器如果你已经掌握了同步迭代器可以进一步了解异步迭代器Async Iterator。Python 3.6 引入了__aiter__()和__anext__()协议配合async for使用。比如用aiohttp异步读取网页内容import aiohttp import asyncio async def fetch_pages(urls): async with aiohttp.ClientSession() as session: for url in urls: async with session.get(url) as resp: yield await resp.text() async def main(): async for content in fetch_pages([https://example.com, https://example.org]): print(len(content)) asyncio.run(main())异步迭代器的思路和同步迭代器完全一致只是取值过程不阻塞线程。理解了同步迭代器和for循环背后的机制再看async for会非常轻松——你只是把原来同步的next()换成了可等待的__anext__()而已。迭代器的应用远不止“省内存”这么简单——它是一种组合数据流的抽象方式是构建可扩展程序的基石。把for循环真正搞明白之后你对 Python 的理解会上升一个台阶写出来的代码也会更接近 Python 的设计哲学简洁、清晰、优雅。如果这个内容对你有所帮助强烈建议你打开终端把上面的代码亲手跑一遍。先写一个最简单的生成器再尝试写一个自定义迭代器类最后用itertools组合出更复杂的数据管道。只有自己写一遍你才能真正理解迭代器为什么是 Python 中最优雅的设计之一。