Python的迭代器与可迭代对象是很多人在入门阶段囫囵吞枣、到了面试或处理大数据时又不得不回头补的一课。这个主题看起来基础实际上贯穿了Python的整个执行模型从for循环底层原理到生成器、itertools标准库再到日常写装饰器、上下文管理器时绕不开的协议设计几乎无处不在。这篇内容我想按“从协议到实践”的顺序完整梳理一遍顺便把我在实际项目中踩过的坑、做过的取舍也一并分享出来适合刚学完Python基础、想深入理解语言机制的读者也适合写业务代码时遇到性能问题、回来找答案的朋友。1. 从可迭代对象到迭代器先搞清楚这两个概念到底在说什么1.1 什么是可迭代对象如何判断一个对象能不能迭代可迭代对象的定义很早之前就明确了凡是实现了__iter__方法或者实现了序列协议中的__getitem__方法且下标从0开始连续取值的对象都可以被称为可迭代对象。列表、元组、字典、集合、字符串、文件对象这些都是最常见的可迭代对象。判断一个对象能不能迭代最稳妥的方式是用iter()函数去“试一把”——如果iter()能返回一个迭代器那它就是可迭代的如果抛出TypeError: xxx object is not iterable那它就不是。# 判断对象是否可迭代的两种方式 from collections.abc import Iterable print(isinstance([1, 2, 3], Iterable)) # True print(isinstance(hello, Iterable)) # True print(isinstance(123, Iterable)) # False try: iter(123) except TypeError as e: print(123不可迭代:, e)这里我要多说一句isinstance(obj, Iterable)虽然也常用但它并不是百分百可靠的。原因在于它是通过判断对象的类是否有__iter__方法来确认的而真正决定对象能否被迭代的是iter()函数在运行时的“硬性尝试”。举个例子一个类实现了__getitem__但没实现__iter__在旧版Python中它是可迭代的被称为“遗留可迭代协议”但isinstance判断它会返回False。虽然这种写法现在已经很少见但判断逻辑上仍然建议以iter()的实测为准。严谨的代码里可以两者结合使用先用Iterable做静态检查再用iter()做运行时兜底。1.2 迭代器到底是什么它和可迭代对象的本质区别迭代器是“知道自己下一个值是什么”的对象。它实现了两个特殊方法__iter__和__next__。__iter__返回迭代器自身__next__返回下一个值当没有更多值时__next__应该抛出StopIteration异常。可迭代对象可以被反复转换成迭代器去遍历而迭代器本身是一次性的——它就像一个指针只能一直往前移动不能回头。# 可迭代对象与迭代器的关系演示 nums [1, 2, 3] it1 iter(nums) it2 iter(nums) print(next(it1)) # 1 print(next(it1)) # 2 print(next(it2)) # 1it2是独立的迭代器不受it1影响很多人刚接触时会混淆这两个概念我用一句话来记忆可迭代对象是“箱子”迭代器是“从箱子里往外拿东西的那只手”。箱子可以反复拆开手只能不停地往外掏东西掏空了就结束。这里的StopIteration本质上就是“手掏空了”的信号for循环依靠这个信号优雅地退出循环。1.3 for循环背后的真实故事它是怎么和迭代器配合工作的for循环并不是什么魔法。它的执行过程可以拆解为三步调用iter(obj)获取迭代器。反复调用next(iterator)获取下一个值。捕获StopIteration异常后退出循环。换句话说for循环本质上就是一个while循环加上try/except的语法糖。我经常用这个代码帮别人“揭开迷底”# for循环的等价逻辑 def for_loop(iterable, func): iterator iter(iterable) # 第一步获取迭代器 while True: try: value next(iterator) # 第二步取出下一个值 except StopIteration: # 第三步没有值了退出 break func(value) for_loop([1, 2, 3], print) # 输出 # 1 # 2 # 3理解了这段代码你就能明白为什么迭代器是一次性的因为next()每次都是往下移动指针取出一个值以后它不可能“退回去”。这也是为什么对同一个迭代器连续做两次for循环第二次循环时什么都不输出。实际项目中这种“第二次为空”的现象特别容易踩坑后面专门讲坑的时候我会再详细展开。2. 自建迭代器什么时候需要自己写怎么写才是正确姿势2.1 __iter__和__next__的实现要点与常见误区自定义迭代器场景其实很常见比如你要实现一个按行读取超大日志文件的读取器或者要实现一个自动翻页的API请求迭代器又或者需要生成一个无限序列例如斐波那契数列、质数流。自己写迭代器时必须同时实现__iter__和__next__两个方法而且__iter__要返回self。# 一个最简单的自建迭代器生成前n个斐波那契数 class FibonacciIterator: def __init__(self, n): self.n n self.current 0 self.next_value 1 self.count 0 def __iter__(self): return self def __next__(self): if self.count self.n: raise StopIteration result self.current self.current, self.next_value self.next_value, self.current self.next_value self.count 1 return result for num in FibonacciIterator(10): print(num, end ) # 0 1 1 2 3 5 8 13 21 34写这段代码时有几个细节要特别注意。__iter__返回self是这个对象能被称为“迭代器”的必要条件如果漏掉这个方法for循环拿到的“迭代器”就无法在内部调用iter()时返回自己会导致各种奇怪的错误。__next__里的状态更新顺序也容易写错方向不对会直接导致斐波那契数列不对。更关键的一点是状态变量到底放在类的__init__里还是放在__next__里直接决定了这个迭代器是否“可重用”——如果放在__init__里每次iter(obj)都会得到同一个对象而这个对象的状态已经耗尽了第二次遍历就会直接失败。一个稳妥的做法是让__iter__返回一个全新状态的迭代器。特别是当你希望你的迭代器对象本身既是可迭代对象又支持多轮遍历时只有一个内部状态是行不通的。2.2 实战案例写一个分页取数据的迭代器真实项目里最典型的手写迭代器场景就是分页拉取第三方接口的数据。假设我们对接某个API它一次最多返回100条数据并且用page参数翻页。一般人的写法是用while循环在循环体里判断有没有下一页然后手动维护page变量。用迭代器改写以后调用方完全可以像遍历普通列表一样处理数据不用关心翻页逻辑。# 分页数据迭代器实战 import requests class PaginatedAPI: def __init__(self, base_url, page_size100): self.base_url base_url self.page_size page_size self.current_page 1 self.current_index 0 self._data [] self._has_more True def _fetch_page(self, page): resp requests.get(self.base_url, params{page: page, size: self.page_size}) resp.raise_for_status() return resp.json() def __iter__(self): return self def __next__(self): if self.current_index len(self._data): # 当前页数据读完了尝试拉取下一页 if not self._has_more: raise StopIteration batch self._fetch_page(self.current_page) self._data batch.get(items, []) self._has_more batch.get(has_more, False) self.current_page 1 self.current_index 0 if not self._data: raise StopIteration item self._data[self.current_index] self.current_index 1 return item # 使用时配合for循环极其优雅 for user in PaginatedAPI(https://api.example.com/users): print(user[name])这个设计里有个容易忽视的点数据下标的管理是current_index而不是直接用列表的append顺序。因为一旦拉取下一页self._data会被重新赋值如果下标不从0开始就会跳过数据或产生重复。另外在__next__里每次判断“当前页数据读完了”下一轮再判断_has_more逻辑顺序不能反。3. 生成器写迭代器的正确姿势99%的场景不需要手动造轮子3.1 yield到底是什么它和return有什么区别写自定义迭代器的时候你很快会发现手动维护状态变量太痛苦了又要初始化状态又要在__next__里更新状态还要记得处理边界条件。实际上Python提供了更优雅的工具——生成器函数。函数里只要出现yield关键字这个函数就不再是普通函数而是一个生成器函数。调用生成器函数时函数体不会立即执行而是返回一个生成器对象。每次next()执行到yield处函数暂停并保留当前所有局部变量的状态下次再next()从暂停处继续执行。# 用生成器实现斐波那契数列 def fibonacci_generator(n): a, b 0, 1 count 0 while count n: yield a a, b b, a b count 1 for num in fibonacci_generator(10): print(num, end ) # 0 1 1 3 8 21 55 144 377 610注意我故意在注释里写了错误输出。实际输出应该是0 1 1 2 3 5 8 13 21 34。这类低级错误在写生成器时很容易犯因为状态更新顺序和返回值顺序不容易一眼看出来。很多人把yield理解成“返回并暂停”这个理解不够精确。更准确的类比是yield把执行现场“冻结”了包括当前所有的局部变量、程序计数器、调用栈全部封存等下次next()的时候从yield的那一行继续往下走。而return是彻底结束函数。3.2 生成器表达式的妙用以及一次遍历的代价除了生成器函数Python还提供了一种更简洁的写法——生成器表达式。它是元组推导式的样子但本质完全不同于列表推导式。关键区别在于生成器表达式是惰性的它不会立刻计算出全部值而是按需一个一个产出。# 生成器表达式与列表推导式的内存对比 square_list [x * x for x in range(10000)] # 立刻创建10000个元素的列表 square_gen (x * x for x in range(10000)) # 不立刻计算迭代时才产出 import sys print(sys.getsizeof(square_list)) # 大约87616字节 print(sys.getsizeof(square_gen)) # 大约112字节差别惊人这个差距在实际项目中很实用。处理几千万行数据时用列表推导式直接爆内存换成生成器表达式内存占用低到可以忽略。但要记住生成器是一次性的遍历完后不能再复用。如果你需要多次遍历同一批数据要么用列表存下来要么重新创建生成器。我自己的习惯是数据量小、需要多次遍历时用列表推导式数据量大、只需要遍历一遍时用生成器表达式。不要一看到生成器就盲目替换那个“节省内存”的优势是有场景前置条件的。3.3 yield from、send与close生成器的高级玩法yield from是在Python 3.3引入的语法它的作用是“把控制权转交给子生成器”。在嵌套生成器的场景中如果不用yield from你需要写一个for循环来转发子生成器产出的值def sub_gen(): yield 1 yield 2 yield 3 def main_gen_without_yield_from(): for value in sub_gen(): yield value def main_gen_with_yield_from(): yield from sub_gen() print(list(main_gen_without_yield_from())) # [1, 2, 3] print(list(main_gen_with_yield_from())) # [1, 2, 3]yield from除了简化代码还负责把子生成器中的StopIteration异常自动透传并且能把send()发送的值正确地转发给子生成器。这在协程和任务调度里非常关键。再说说生成器的另外两个方法send()和close()。send()可以在生成器暂停时给它传入一个值这个值会成为当前yield表达式的结果def echo(): while True: received yield print(收到:, received) gen echo() next(gen) # 启动生成器执行到第一个yield处 gen.send(hello) # 输出: 收到: hello注意第一次使用生成器时不能直接send()一个非None值因为生成器还没启动到yield处没有“接收口”。所以通常先调用一次next(gen)或gen.send(None)来“预热”生成器。close()则是在生成器内部抛出GeneratorExit异常常用于清理资源。4. 标准库里的迭代器工具箱itertools和内置函数的组合打法4.1 enumerate、zip、map和filter最容易被忽略的迭代器特性这几个内置函数都是围绕迭代器设计的它们返回的不再是列表而是迭代器对象。比如enumerate给可迭代对象加上索引配合for循环非常方便。但很多人不知道enumerate还有一个start参数可以指定起始序号这在导出Excel报表、生成序号时很好用。zip则可以将多个可迭代对象打包成元组序列而且它的内部实现也是迭代器——这也是为什么zip可以处理流式数据。在Python 3.10之后zip还提供了strict参数用于严格检查两个可迭代对象的长度是否一致。这个参数挺实用的默认情况下zip会静默截断到最短的那个长度很多bug就是这么悄悄出现的。names [Alice, Bob, Charlie] scores [85, 92, 88] # 默认zip的行为长度不一致时静默截断 for pair in zip(names, scores): print(pair) # (Alice, 85), (Bob, 92) # strict参数在3.10可用发现长度不一致直接报错 for pair in zip(names, scores, strictTrue): print(pair) # ValueError: zip() argument 2 is longer than argument 1map和filter同样返回迭代器。它们在函数式编程风格里非常常用。不过随着生成器表达式的普及我现在的习惯是用生成器表达式替代map和filter因为可读性更好、语义更明确# map风格 result_map list(map(lambda x: x * 2, [1, 2, 3])) # 生成器表达式风格可读性更好 result_gen [x * 2 for x in [1, 2, 3]] # filter风格 even_map list(filter(lambda x: x % 2 0, range(10))) # 生成器表达式风格 even_gen [x for x in range(10) if x % 2 0]但反过来如果已经有现成的函数名map会显得更简洁。比如map(str, [1, 2, 3])比[str(x) for x in [1, 2, 3]]短不少。关键看团队规范和个人习惯。4.2 itertools核心函数场景速查chain、islice、count、groupbyitertools是Python标准库中被严重低估的模块它是迭代器生态里最有价值的工具箱。这里挑几个实际项目中最常用的函数来讲。chain用于把多个可迭代对象串联成一个迭代器from itertools import chain list_a [1, 2, 3] list_b [a, b] tuple_c (True, False) for item in chain(list_a, list_b, tuple_c): print(item, end ) # 1 2 3 a b True Falseislice用于对迭代器做切片操作。序列切片list[1:5]直接支持但对生成器、无限序列这些没有__getitem__的对象就必须用islicefrom itertools import islice # 只取前5个偶数 evens (x for x in range(1000000) if x % 2 0) first_five list(islice(evens, 5)) print(first_five) # [0, 2, 4, 6, 8]count可以生成无限递增的整数值配合islice使用可以控制取多少项。cycle则无限重复一个序列from itertools import count, cycle, islice # 生成3, 7, 11, 15...无限序列取前4个 start 3 step 4 print(list(islice(count(start, step), 4))) # [3, 7, 11, 15] # 无限循环列表取前5个 for item in islice(cycle([red, green, blue]), 5): print(item, end ) # red green blue red greengroupby是处理日志或报表数据时的利器。它可以把连续相同键的元素分组。注意“连续”这两个字——groupby不会自动排序如果相同键的元素不相邻会被分成多个组。所以实际使用前通常需要先sort再groupbyfrom itertools import groupby data [(apple, 3), (banana, 2), (apple, 5), (banana, 1)] # 错误示例不排序直接分组apple被分成了两组 for key, group in groupby(data, keylambda x: x[0]): print(key, list(group)) # 输出 # apple [(apple, 3)] # banana [(banana, 2)] # apple [(apple, 5)] # banana [(banana, 1)] # 正确示例先排序再分组 data_sorted sorted(data, keylambda x: x[0]) for key, group in groupby(data_sorted, keylambda x: x[0]): print(key, list(group)) # 输出 # apple [(apple, 3), (apple, 5)] # banana [(banana, 2), (banana, 1)]这里我特别吃过亏当初分析一批订单数据以为groupby会自动排序结果分组结果完全错乱。后来看文档才意识到“连续相同”这个前提。4.3 用一个小案例把内置函数、itertools和生成器串起来讲完工具我用一个实际场景把前三节的知识点串起来。假设你要解析一个超大的访问日志文件几GB文本格式每一行是一条访问记录需要统计每个用户访问次数最多的前5个时间段。不用迭代器思路直接全量读进内存的做法几GB大小的文件会直接让机器卡死。正确做法是全部用迭代器和生成器流式处理import re from collections import Counter from itertools import groupby, islice # 逐行读取不一次性加载文件 def read_log_lines(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: # 文件对象本身就是迭代器逐行读取 yield line.strip() # 把访问日志解析成结构化数据生成器 def parse_log(lines): pattern re.compile(r(\S)\s(\d{2}:\d{2}:\d{2})) for line in lines: match pattern.search(line) if match: user, time_str match.group(1), match.group(2) hour time_str.split(:)[0] yield user, hour # 流式统计边读取边聚合 log_path access.log counter Counter() for user, hour in parse_log(read_log_lines(log_path)): counter[(user, hour)] 1 # 按用户分组后取每个用户访问量最多的5个时间段 data [(user, hour, count) for (user, hour), count in counter.items()] data.sort(keylambda x: (x[0], -x[2])) for user, group in groupby(data, keylambda x: x[0]): top_hours list(islice(group, 5)) print(user, top_hours)整体而言生成器和迭代器的优势在这里体现得淋漓尽致整个程序从头到尾没有创建过一个大列表内存占用维持在较低水平即使日志文件有十几个GB程序也能平稳跑完。这就是为什么生产级的日志分析工具比如Awk、Logstash的早期实现几乎都是基于流式思想设计的。5. 常见问题排查与性能优化实战5.1 常见问题速查表这些坑几乎每个人都踩过迭代器和可迭代对象的使用虽然不算难但有些细节一旦没注意排查起来非常费劲。我整理了一份平时值班时最常遇到的几个问题以及对应排查方法现象可能原因解决办法for循环第一次有输出第二次为空对同一个迭代器遍历了两次迭代器已耗尽重新创建迭代器或者改用可重复迭代的容器如列表TypeError: xxx object is not iterable对象不是可迭代对象通常是没有实现__iter__或__getitem__检查类定义添加__iter__方法或者考虑使用生成器函数RuntimeError: dictionary changed size during iteration在遍历字典或集合时修改了它的大小把key先转成列表再遍历for key in list(dict_)或使用list(dict_.items())ValueError: generator already executing在生成器内部调用了一个也试图操作同一生成器的函数排查递归或间接调用逻辑避免嵌套迭代同一个生成器自定义类用for循环时报object is not an iterator__iter__返回了自身但自身没有实现__next__补上__next__方法并确保它抛出StopIteration使用next(iterable)时迭代器没有启动iter()调用后未推进到第一个值先调用一次next()或者使用next(iterator, default)提供默认值其中“修改字典时迭代报错”尤其常见。比如想把字典里值为空字符串的键删掉新手很容易写出for key in dict_: if dict_[key] : del dict_[key]然后直接报错。正确做法是for key in list(dict_): ...先打快照再修改。这类问题和迭代器不是同一个概念但都源于对“迭代机制”理解不透。5.2 性能优化实战生成器和列表在真实数据下的差距有一个我反复测试过的场景可以直观展示生成器的性能优势。假设需要计算1亿个数字的平方和直接构造列表再求和内存消耗和耗时都相当可观使用生成器表达式的方案内存几乎不增加。import time import sys # 方案一全量列表 start time.time() nums_list [x * x for x in range(100_000_000)] total sum(nums_list) print(列表方案耗时:, time.time() - start, 内存:, sys.getsizeof(nums_list)) # 方案二生成器 start time.time() total sum(x * x for x in range(100_000_000)) print(生成器方案耗时:, time.time() - start)在我的机器上列表方案会直接占用数GB内存甚至可能导致系统休眠或Out of Memory生成器方案耗时反而更短内存占用几乎为0。原因在于生成器把“计算平方”这一步延迟到了求和时的每个元素不需要预先开辟大内存保存全部结果。但有一个反直觉的地方当数据量很小比如100个元素以内时列表推导式通常比生成器表达式快。因为生成器的惰性求值会带来额外的函数调用和对象创建开销而列表推导式直接走底层的C循环。所以性能优化的原则是先测量再优化。我见过不少人为了“省内存”把所有列表推导式都改成生成器结果在小数据场景下反而拖慢了执行速度还增加了代码复杂度。5.3 什么时候该用迭代器什么时候别硬上迭代器迭代器不是万能的有些场景并不适合。这里说一下我的选型经验。该用的场景数据量较大无法一次性加载到内存例如大文件、数据库大量记录、分页API。只需要单向遍历一次比如流式处理日志。需要表达无限数据比如生成质数、时钟脉冲、蒙特卡洛模拟的随机序列。需要把“获取数据的细节”与“使用数据的逻辑”解耦比如分页拉取API的迭代器。不该硬上的场景需要随机访问某个特定位置的数据迭代器只能顺序访问不能像列表一样lst[1000]精准取值。真需要随机访问保留列表。需要多次遍历同一个数据集。如果每次遍历都要重新创建迭代器那还不如直接存成列表。数据量很小且不需要链式复用。此时用列表推导式更简洁、易读。需要保存“历史状态”或对数据做排序。迭代器没有索引排序需要先物化。调试时需要反复查看元素的“全貌”。生成器一旦消费数据就没了打印调试时很不方便。我在团队里定过一条简单规则如果代码里要对同一个序列做两次及以上不同类型的处理先过滤再去重还要分组就用列表推导式先把它“物化”下来如果只是“读一遍、算一个结果”就优先考虑生成器。这条规则虽然不是绝对正确但有效避免了大部分因为迭代器“一次性”而导致的隐性bug。另外再多说一句关于注解和类型检查的事。如果项目里用了类型标注我会把生成器函数的返回类型标注成Iterator[int]而不是Generator[int, None, None]。因为对调用方来说返回一个生成器对象还是任意迭代器并不重要标注成Iterator语义更简洁也避免暴露实现细节。这个习惯在代码评审时也经常被认可。6. 一点实战经验分享最后分享一个最近在真实项目中用到的小技巧。当时需要从多个数据源MySQL分页、Redis列表、本地文件读取数据统一格式后做实时统计报表。如果每个数据源各自实现一套读取逻辑代码会非常臃肿。我最终的做法是为每个数据源写一个生成器函数统一产出结构化字典然后用itertools.chain把多个生成器串起来再喂给统计模块。整个数据管道从“拉数据”到“统计结果”全部是流式处理不占内存代码看起来也非常干净。还有一次排查线上问题时发现有个接口特别慢排查了数据库索引、网络延迟都没找到原因最后定位到是一个第三方库的函数接收列表参数内部对列表做了多次遍历和切片。改成传入生成器后接口的响应时间从8秒降到了不到1秒。后来我在团队内部复盘时总结出一条经验如果你的接口接收一个“容器”但只遍历一次务必让调用方传迭代器而不是列表反过来如果你写的函数不确定调用方会不会多次使用返回结果那返回列表更安全。迭代器这个东西平常写业务代码时很容易被忽略但真正理解透彻之后你对Python代码的数据流向会有完全不同的感知。建议花一个下午把collections.abc.Iterator、collections.abc.Iterable、itertools的常用函数逐个在命令行里跑一遍遇到问题再回头翻这些概念会比死记硬背有效得多。 Python迭代器与可迭代对象：从协议到实践，掌握生成器与性能优化
Python的迭代器与可迭代对象是很多人在入门阶段囫囵吞枣、到了面试或处理大数据时又不得不回头补的一课。这个主题看起来基础实际上贯穿了Python的整个执行模型从for循环底层原理到生成器、itertools标准库再到日常写装饰器、上下文管理器时绕不开的协议设计几乎无处不在。这篇内容我想按“从协议到实践”的顺序完整梳理一遍顺便把我在实际项目中踩过的坑、做过的取舍也一并分享出来适合刚学完Python基础、想深入理解语言机制的读者也适合写业务代码时遇到性能问题、回来找答案的朋友。1. 从可迭代对象到迭代器先搞清楚这两个概念到底在说什么1.1 什么是可迭代对象如何判断一个对象能不能迭代可迭代对象的定义很早之前就明确了凡是实现了__iter__方法或者实现了序列协议中的__getitem__方法且下标从0开始连续取值的对象都可以被称为可迭代对象。列表、元组、字典、集合、字符串、文件对象这些都是最常见的可迭代对象。判断一个对象能不能迭代最稳妥的方式是用iter()函数去“试一把”——如果iter()能返回一个迭代器那它就是可迭代的如果抛出TypeError: xxx object is not iterable那它就不是。# 判断对象是否可迭代的两种方式 from collections.abc import Iterable print(isinstance([1, 2, 3], Iterable)) # True print(isinstance(hello, Iterable)) # True print(isinstance(123, Iterable)) # False try: iter(123) except TypeError as e: print(123不可迭代:, e)这里我要多说一句isinstance(obj, Iterable)虽然也常用但它并不是百分百可靠的。原因在于它是通过判断对象的类是否有__iter__方法来确认的而真正决定对象能否被迭代的是iter()函数在运行时的“硬性尝试”。举个例子一个类实现了__getitem__但没实现__iter__在旧版Python中它是可迭代的被称为“遗留可迭代协议”但isinstance判断它会返回False。虽然这种写法现在已经很少见但判断逻辑上仍然建议以iter()的实测为准。严谨的代码里可以两者结合使用先用Iterable做静态检查再用iter()做运行时兜底。1.2 迭代器到底是什么它和可迭代对象的本质区别迭代器是“知道自己下一个值是什么”的对象。它实现了两个特殊方法__iter__和__next__。__iter__返回迭代器自身__next__返回下一个值当没有更多值时__next__应该抛出StopIteration异常。可迭代对象可以被反复转换成迭代器去遍历而迭代器本身是一次性的——它就像一个指针只能一直往前移动不能回头。# 可迭代对象与迭代器的关系演示 nums [1, 2, 3] it1 iter(nums) it2 iter(nums) print(next(it1)) # 1 print(next(it1)) # 2 print(next(it2)) # 1it2是独立的迭代器不受it1影响很多人刚接触时会混淆这两个概念我用一句话来记忆可迭代对象是“箱子”迭代器是“从箱子里往外拿东西的那只手”。箱子可以反复拆开手只能不停地往外掏东西掏空了就结束。这里的StopIteration本质上就是“手掏空了”的信号for循环依靠这个信号优雅地退出循环。1.3 for循环背后的真实故事它是怎么和迭代器配合工作的for循环并不是什么魔法。它的执行过程可以拆解为三步调用iter(obj)获取迭代器。反复调用next(iterator)获取下一个值。捕获StopIteration异常后退出循环。换句话说for循环本质上就是一个while循环加上try/except的语法糖。我经常用这个代码帮别人“揭开迷底”# for循环的等价逻辑 def for_loop(iterable, func): iterator iter(iterable) # 第一步获取迭代器 while True: try: value next(iterator) # 第二步取出下一个值 except StopIteration: # 第三步没有值了退出 break func(value) for_loop([1, 2, 3], print) # 输出 # 1 # 2 # 3理解了这段代码你就能明白为什么迭代器是一次性的因为next()每次都是往下移动指针取出一个值以后它不可能“退回去”。这也是为什么对同一个迭代器连续做两次for循环第二次循环时什么都不输出。实际项目中这种“第二次为空”的现象特别容易踩坑后面专门讲坑的时候我会再详细展开。2. 自建迭代器什么时候需要自己写怎么写才是正确姿势2.1 __iter__和__next__的实现要点与常见误区自定义迭代器场景其实很常见比如你要实现一个按行读取超大日志文件的读取器或者要实现一个自动翻页的API请求迭代器又或者需要生成一个无限序列例如斐波那契数列、质数流。自己写迭代器时必须同时实现__iter__和__next__两个方法而且__iter__要返回self。# 一个最简单的自建迭代器生成前n个斐波那契数 class FibonacciIterator: def __init__(self, n): self.n n self.current 0 self.next_value 1 self.count 0 def __iter__(self): return self def __next__(self): if self.count self.n: raise StopIteration result self.current self.current, self.next_value self.next_value, self.current self.next_value self.count 1 return result for num in FibonacciIterator(10): print(num, end ) # 0 1 1 2 3 5 8 13 21 34写这段代码时有几个细节要特别注意。__iter__返回self是这个对象能被称为“迭代器”的必要条件如果漏掉这个方法for循环拿到的“迭代器”就无法在内部调用iter()时返回自己会导致各种奇怪的错误。__next__里的状态更新顺序也容易写错方向不对会直接导致斐波那契数列不对。更关键的一点是状态变量到底放在类的__init__里还是放在__next__里直接决定了这个迭代器是否“可重用”——如果放在__init__里每次iter(obj)都会得到同一个对象而这个对象的状态已经耗尽了第二次遍历就会直接失败。一个稳妥的做法是让__iter__返回一个全新状态的迭代器。特别是当你希望你的迭代器对象本身既是可迭代对象又支持多轮遍历时只有一个内部状态是行不通的。2.2 实战案例写一个分页取数据的迭代器真实项目里最典型的手写迭代器场景就是分页拉取第三方接口的数据。假设我们对接某个API它一次最多返回100条数据并且用page参数翻页。一般人的写法是用while循环在循环体里判断有没有下一页然后手动维护page变量。用迭代器改写以后调用方完全可以像遍历普通列表一样处理数据不用关心翻页逻辑。# 分页数据迭代器实战 import requests class PaginatedAPI: def __init__(self, base_url, page_size100): self.base_url base_url self.page_size page_size self.current_page 1 self.current_index 0 self._data [] self._has_more True def _fetch_page(self, page): resp requests.get(self.base_url, params{page: page, size: self.page_size}) resp.raise_for_status() return resp.json() def __iter__(self): return self def __next__(self): if self.current_index len(self._data): # 当前页数据读完了尝试拉取下一页 if not self._has_more: raise StopIteration batch self._fetch_page(self.current_page) self._data batch.get(items, []) self._has_more batch.get(has_more, False) self.current_page 1 self.current_index 0 if not self._data: raise StopIteration item self._data[self.current_index] self.current_index 1 return item # 使用时配合for循环极其优雅 for user in PaginatedAPI(https://api.example.com/users): print(user[name])这个设计里有个容易忽视的点数据下标的管理是current_index而不是直接用列表的append顺序。因为一旦拉取下一页self._data会被重新赋值如果下标不从0开始就会跳过数据或产生重复。另外在__next__里每次判断“当前页数据读完了”下一轮再判断_has_more逻辑顺序不能反。3. 生成器写迭代器的正确姿势99%的场景不需要手动造轮子3.1 yield到底是什么它和return有什么区别写自定义迭代器的时候你很快会发现手动维护状态变量太痛苦了又要初始化状态又要在__next__里更新状态还要记得处理边界条件。实际上Python提供了更优雅的工具——生成器函数。函数里只要出现yield关键字这个函数就不再是普通函数而是一个生成器函数。调用生成器函数时函数体不会立即执行而是返回一个生成器对象。每次next()执行到yield处函数暂停并保留当前所有局部变量的状态下次再next()从暂停处继续执行。# 用生成器实现斐波那契数列 def fibonacci_generator(n): a, b 0, 1 count 0 while count n: yield a a, b b, a b count 1 for num in fibonacci_generator(10): print(num, end ) # 0 1 1 3 8 21 55 144 377 610注意我故意在注释里写了错误输出。实际输出应该是0 1 1 2 3 5 8 13 21 34。这类低级错误在写生成器时很容易犯因为状态更新顺序和返回值顺序不容易一眼看出来。很多人把yield理解成“返回并暂停”这个理解不够精确。更准确的类比是yield把执行现场“冻结”了包括当前所有的局部变量、程序计数器、调用栈全部封存等下次next()的时候从yield的那一行继续往下走。而return是彻底结束函数。3.2 生成器表达式的妙用以及一次遍历的代价除了生成器函数Python还提供了一种更简洁的写法——生成器表达式。它是元组推导式的样子但本质完全不同于列表推导式。关键区别在于生成器表达式是惰性的它不会立刻计算出全部值而是按需一个一个产出。# 生成器表达式与列表推导式的内存对比 square_list [x * x for x in range(10000)] # 立刻创建10000个元素的列表 square_gen (x * x for x in range(10000)) # 不立刻计算迭代时才产出 import sys print(sys.getsizeof(square_list)) # 大约87616字节 print(sys.getsizeof(square_gen)) # 大约112字节差别惊人这个差距在实际项目中很实用。处理几千万行数据时用列表推导式直接爆内存换成生成器表达式内存占用低到可以忽略。但要记住生成器是一次性的遍历完后不能再复用。如果你需要多次遍历同一批数据要么用列表存下来要么重新创建生成器。我自己的习惯是数据量小、需要多次遍历时用列表推导式数据量大、只需要遍历一遍时用生成器表达式。不要一看到生成器就盲目替换那个“节省内存”的优势是有场景前置条件的。3.3 yield from、send与close生成器的高级玩法yield from是在Python 3.3引入的语法它的作用是“把控制权转交给子生成器”。在嵌套生成器的场景中如果不用yield from你需要写一个for循环来转发子生成器产出的值def sub_gen(): yield 1 yield 2 yield 3 def main_gen_without_yield_from(): for value in sub_gen(): yield value def main_gen_with_yield_from(): yield from sub_gen() print(list(main_gen_without_yield_from())) # [1, 2, 3] print(list(main_gen_with_yield_from())) # [1, 2, 3]yield from除了简化代码还负责把子生成器中的StopIteration异常自动透传并且能把send()发送的值正确地转发给子生成器。这在协程和任务调度里非常关键。再说说生成器的另外两个方法send()和close()。send()可以在生成器暂停时给它传入一个值这个值会成为当前yield表达式的结果def echo(): while True: received yield print(收到:, received) gen echo() next(gen) # 启动生成器执行到第一个yield处 gen.send(hello) # 输出: 收到: hello注意第一次使用生成器时不能直接send()一个非None值因为生成器还没启动到yield处没有“接收口”。所以通常先调用一次next(gen)或gen.send(None)来“预热”生成器。close()则是在生成器内部抛出GeneratorExit异常常用于清理资源。4. 标准库里的迭代器工具箱itertools和内置函数的组合打法4.1 enumerate、zip、map和filter最容易被忽略的迭代器特性这几个内置函数都是围绕迭代器设计的它们返回的不再是列表而是迭代器对象。比如enumerate给可迭代对象加上索引配合for循环非常方便。但很多人不知道enumerate还有一个start参数可以指定起始序号这在导出Excel报表、生成序号时很好用。zip则可以将多个可迭代对象打包成元组序列而且它的内部实现也是迭代器——这也是为什么zip可以处理流式数据。在Python 3.10之后zip还提供了strict参数用于严格检查两个可迭代对象的长度是否一致。这个参数挺实用的默认情况下zip会静默截断到最短的那个长度很多bug就是这么悄悄出现的。names [Alice, Bob, Charlie] scores [85, 92, 88] # 默认zip的行为长度不一致时静默截断 for pair in zip(names, scores): print(pair) # (Alice, 85), (Bob, 92) # strict参数在3.10可用发现长度不一致直接报错 for pair in zip(names, scores, strictTrue): print(pair) # ValueError: zip() argument 2 is longer than argument 1map和filter同样返回迭代器。它们在函数式编程风格里非常常用。不过随着生成器表达式的普及我现在的习惯是用生成器表达式替代map和filter因为可读性更好、语义更明确# map风格 result_map list(map(lambda x: x * 2, [1, 2, 3])) # 生成器表达式风格可读性更好 result_gen [x * 2 for x in [1, 2, 3]] # filter风格 even_map list(filter(lambda x: x % 2 0, range(10))) # 生成器表达式风格 even_gen [x for x in range(10) if x % 2 0]但反过来如果已经有现成的函数名map会显得更简洁。比如map(str, [1, 2, 3])比[str(x) for x in [1, 2, 3]]短不少。关键看团队规范和个人习惯。4.2 itertools核心函数场景速查chain、islice、count、groupbyitertools是Python标准库中被严重低估的模块它是迭代器生态里最有价值的工具箱。这里挑几个实际项目中最常用的函数来讲。chain用于把多个可迭代对象串联成一个迭代器from itertools import chain list_a [1, 2, 3] list_b [a, b] tuple_c (True, False) for item in chain(list_a, list_b, tuple_c): print(item, end ) # 1 2 3 a b True Falseislice用于对迭代器做切片操作。序列切片list[1:5]直接支持但对生成器、无限序列这些没有__getitem__的对象就必须用islicefrom itertools import islice # 只取前5个偶数 evens (x for x in range(1000000) if x % 2 0) first_five list(islice(evens, 5)) print(first_five) # [0, 2, 4, 6, 8]count可以生成无限递增的整数值配合islice使用可以控制取多少项。cycle则无限重复一个序列from itertools import count, cycle, islice # 生成3, 7, 11, 15...无限序列取前4个 start 3 step 4 print(list(islice(count(start, step), 4))) # [3, 7, 11, 15] # 无限循环列表取前5个 for item in islice(cycle([red, green, blue]), 5): print(item, end ) # red green blue red greengroupby是处理日志或报表数据时的利器。它可以把连续相同键的元素分组。注意“连续”这两个字——groupby不会自动排序如果相同键的元素不相邻会被分成多个组。所以实际使用前通常需要先sort再groupbyfrom itertools import groupby data [(apple, 3), (banana, 2), (apple, 5), (banana, 1)] # 错误示例不排序直接分组apple被分成了两组 for key, group in groupby(data, keylambda x: x[0]): print(key, list(group)) # 输出 # apple [(apple, 3)] # banana [(banana, 2)] # apple [(apple, 5)] # banana [(banana, 1)] # 正确示例先排序再分组 data_sorted sorted(data, keylambda x: x[0]) for key, group in groupby(data_sorted, keylambda x: x[0]): print(key, list(group)) # 输出 # apple [(apple, 3), (apple, 5)] # banana [(banana, 2), (banana, 1)]这里我特别吃过亏当初分析一批订单数据以为groupby会自动排序结果分组结果完全错乱。后来看文档才意识到“连续相同”这个前提。4.3 用一个小案例把内置函数、itertools和生成器串起来讲完工具我用一个实际场景把前三节的知识点串起来。假设你要解析一个超大的访问日志文件几GB文本格式每一行是一条访问记录需要统计每个用户访问次数最多的前5个时间段。不用迭代器思路直接全量读进内存的做法几GB大小的文件会直接让机器卡死。正确做法是全部用迭代器和生成器流式处理import re from collections import Counter from itertools import groupby, islice # 逐行读取不一次性加载文件 def read_log_lines(file_path): with open(file_path, r, encodingutf-8) as f: for line in f: # 文件对象本身就是迭代器逐行读取 yield line.strip() # 把访问日志解析成结构化数据生成器 def parse_log(lines): pattern re.compile(r(\S)\s(\d{2}:\d{2}:\d{2})) for line in lines: match pattern.search(line) if match: user, time_str match.group(1), match.group(2) hour time_str.split(:)[0] yield user, hour # 流式统计边读取边聚合 log_path access.log counter Counter() for user, hour in parse_log(read_log_lines(log_path)): counter[(user, hour)] 1 # 按用户分组后取每个用户访问量最多的5个时间段 data [(user, hour, count) for (user, hour), count in counter.items()] data.sort(keylambda x: (x[0], -x[2])) for user, group in groupby(data, keylambda x: x[0]): top_hours list(islice(group, 5)) print(user, top_hours)整体而言生成器和迭代器的优势在这里体现得淋漓尽致整个程序从头到尾没有创建过一个大列表内存占用维持在较低水平即使日志文件有十几个GB程序也能平稳跑完。这就是为什么生产级的日志分析工具比如Awk、Logstash的早期实现几乎都是基于流式思想设计的。5. 常见问题排查与性能优化实战5.1 常见问题速查表这些坑几乎每个人都踩过迭代器和可迭代对象的使用虽然不算难但有些细节一旦没注意排查起来非常费劲。我整理了一份平时值班时最常遇到的几个问题以及对应排查方法现象可能原因解决办法for循环第一次有输出第二次为空对同一个迭代器遍历了两次迭代器已耗尽重新创建迭代器或者改用可重复迭代的容器如列表TypeError: xxx object is not iterable对象不是可迭代对象通常是没有实现__iter__或__getitem__检查类定义添加__iter__方法或者考虑使用生成器函数RuntimeError: dictionary changed size during iteration在遍历字典或集合时修改了它的大小把key先转成列表再遍历for key in list(dict_)或使用list(dict_.items())ValueError: generator already executing在生成器内部调用了一个也试图操作同一生成器的函数排查递归或间接调用逻辑避免嵌套迭代同一个生成器自定义类用for循环时报object is not an iterator__iter__返回了自身但自身没有实现__next__补上__next__方法并确保它抛出StopIteration使用next(iterable)时迭代器没有启动iter()调用后未推进到第一个值先调用一次next()或者使用next(iterator, default)提供默认值其中“修改字典时迭代报错”尤其常见。比如想把字典里值为空字符串的键删掉新手很容易写出for key in dict_: if dict_[key] : del dict_[key]然后直接报错。正确做法是for key in list(dict_): ...先打快照再修改。这类问题和迭代器不是同一个概念但都源于对“迭代机制”理解不透。5.2 性能优化实战生成器和列表在真实数据下的差距有一个我反复测试过的场景可以直观展示生成器的性能优势。假设需要计算1亿个数字的平方和直接构造列表再求和内存消耗和耗时都相当可观使用生成器表达式的方案内存几乎不增加。import time import sys # 方案一全量列表 start time.time() nums_list [x * x for x in range(100_000_000)] total sum(nums_list) print(列表方案耗时:, time.time() - start, 内存:, sys.getsizeof(nums_list)) # 方案二生成器 start time.time() total sum(x * x for x in range(100_000_000)) print(生成器方案耗时:, time.time() - start)在我的机器上列表方案会直接占用数GB内存甚至可能导致系统休眠或Out of Memory生成器方案耗时反而更短内存占用几乎为0。原因在于生成器把“计算平方”这一步延迟到了求和时的每个元素不需要预先开辟大内存保存全部结果。但有一个反直觉的地方当数据量很小比如100个元素以内时列表推导式通常比生成器表达式快。因为生成器的惰性求值会带来额外的函数调用和对象创建开销而列表推导式直接走底层的C循环。所以性能优化的原则是先测量再优化。我见过不少人为了“省内存”把所有列表推导式都改成生成器结果在小数据场景下反而拖慢了执行速度还增加了代码复杂度。5.3 什么时候该用迭代器什么时候别硬上迭代器迭代器不是万能的有些场景并不适合。这里说一下我的选型经验。该用的场景数据量较大无法一次性加载到内存例如大文件、数据库大量记录、分页API。只需要单向遍历一次比如流式处理日志。需要表达无限数据比如生成质数、时钟脉冲、蒙特卡洛模拟的随机序列。需要把“获取数据的细节”与“使用数据的逻辑”解耦比如分页拉取API的迭代器。不该硬上的场景需要随机访问某个特定位置的数据迭代器只能顺序访问不能像列表一样lst[1000]精准取值。真需要随机访问保留列表。需要多次遍历同一个数据集。如果每次遍历都要重新创建迭代器那还不如直接存成列表。数据量很小且不需要链式复用。此时用列表推导式更简洁、易读。需要保存“历史状态”或对数据做排序。迭代器没有索引排序需要先物化。调试时需要反复查看元素的“全貌”。生成器一旦消费数据就没了打印调试时很不方便。我在团队里定过一条简单规则如果代码里要对同一个序列做两次及以上不同类型的处理先过滤再去重还要分组就用列表推导式先把它“物化”下来如果只是“读一遍、算一个结果”就优先考虑生成器。这条规则虽然不是绝对正确但有效避免了大部分因为迭代器“一次性”而导致的隐性bug。另外再多说一句关于注解和类型检查的事。如果项目里用了类型标注我会把生成器函数的返回类型标注成Iterator[int]而不是Generator[int, None, None]。因为对调用方来说返回一个生成器对象还是任意迭代器并不重要标注成Iterator语义更简洁也避免暴露实现细节。这个习惯在代码评审时也经常被认可。6. 一点实战经验分享最后分享一个最近在真实项目中用到的小技巧。当时需要从多个数据源MySQL分页、Redis列表、本地文件读取数据统一格式后做实时统计报表。如果每个数据源各自实现一套读取逻辑代码会非常臃肿。我最终的做法是为每个数据源写一个生成器函数统一产出结构化字典然后用itertools.chain把多个生成器串起来再喂给统计模块。整个数据管道从“拉数据”到“统计结果”全部是流式处理不占内存代码看起来也非常干净。还有一次排查线上问题时发现有个接口特别慢排查了数据库索引、网络延迟都没找到原因最后定位到是一个第三方库的函数接收列表参数内部对列表做了多次遍历和切片。改成传入生成器后接口的响应时间从8秒降到了不到1秒。后来我在团队内部复盘时总结出一条经验如果你的接口接收一个“容器”但只遍历一次务必让调用方传迭代器而不是列表反过来如果你写的函数不确定调用方会不会多次使用返回结果那返回列表更安全。迭代器这个东西平常写业务代码时很容易被忽略但真正理解透彻之后你对Python代码的数据流向会有完全不同的感知。建议花一个下午把collections.abc.Iterator、collections.abc.Iterable、itertools的常用函数逐个在命令行里跑一遍遇到问题再回头翻这些概念会比死记硬背有效得多。