Python迭代器与生成器：从for循环到底层机制完全解析

1. 为什么迭代器是绕不过去的坎三个真实场景先抛一个反直觉的问题range(100)到底算不算迭代器写了两三年 Python 的人十有八九会在这个问题上卡壳。不是大家不努力而是网络上讲迭代器的教程大多停留在“迭代器就是有__next__方法的对象”这个层面背完就忘遇到实际代码照样懵。我在带团队和面试候选人的过程中发现迭代器这个知识点几乎每次都会成为分水岭。问“列表和元组有什么区别”人人都会一问“for 循环是怎么把列表里的元素一个个取出来的”一半人开始含糊。这不怪大家因为迭代器这个机制藏在语言底层平时写业务根本碰不到一旦碰到又往往是在最不该出问题的时候出问题。我先说三个真实场景你感受一下它到底重不重要。1.1 场景一面试官的问题链面试官问“Python 里 for 循环的工作原理是什么”如果你只会回答“就是遍历一个可迭代对象”那等于没答。真正完整的问题链是这样的什么样的对象可以被 for 循环字符串、列表、字典、集合、文件对象它们都能被 for它们是迭代器吗iter()函数做了什么next()函数做了什么为什么对列表调用iter()返回的不是列表本身而是list_iterator自定义一个类怎么让它支持 for 循环两种做法分别是什么这一串问题每一个都指向迭代协议本身。面试官不是想刁难你而是想通过这个问题判断你对自己天天使用的语言有没有底层层面的理解。能答上来的说明代码出了问题有能力往深处挖答不上来的遇到诡异 bug 大概率只能靠 print 大法碰运气。1.2 场景二框架源码里的迭代器模式你用过 Django 的QuerySet吧for user in User.objects.all()这个写法看起来就是遍历一个列表实际上QuerySet是个惰性求值的可迭代对象它直到被真正迭代时才去查数据库而且每迭代一次可能只取一批数据。如果你不理解迭代器的惰性特征你就永远想不通为什么明明查了数据库logger 里却看不到 SQL。再看爬虫场景。你用 requests 拿一个大响应response.iter_content(chunk_size1024)返回的是一个生成器边下载边处理不会把整个文件一次性读进内存。想写出这种高效代码不掌握迭代器和生成器连 API 都看不懂。1.3 场景三海量数据处理时的内存危机这是我最常举的例子。给你一个 5GB 的日志文件要把里面包含“ERROR”的行全部找出来。新手的第一反应是with open(app.log, r, encodingutf-8) as f: lines f.readlines() # 5GB 直接塞进内存程序当场崩掉老手会这样写with open(app.log, r, encodingutf-8) as f: for line in f: # 文件对象本身就是迭代器逐行读取 if ERROR in line: handle(line)同样一个任务第一种写法内存占用好几个 GB第二种写法内存占用稳定在几 KB 到几 MB。区别在哪儿就在文件对象是否以迭代器的方式工作。这不是炫技这是能不能接住任务的差距。理解了这三个场景你应该能认同迭代器不是面试八股它是 Python 里面向数据流编程的基础设施。接下来我带你把它彻底吃透。2. 先把概念掰开揉碎别再把可迭代对象和迭代器混为一谈我见过太多人把“可迭代对象”和“迭代器”当成一回事这是所有混乱的根源。实际上它们是两个不同的概念关系是迭代器一定是可迭代对象但可迭代对象不一定是迭代器。这句话你品一下然后看下面的拆解。2.1 可迭代对象是可以拿去循环的东西可迭代对象Iterable指的是可以被 for 循环遍历的对象。列表、元组、字符串、字典、集合、文件对象、range()的返回值这些都是可迭代对象。判断一个对象是不是可迭代对象最直接的方法是用iter()函数iter([1, 2, 3]) # list_iterator object at 0x... iter(hello) # str_iterator object at 0x... iter(123) # TypeError: int object is not iterable能传给iter()并成功返回的就是可迭代对象。iter()函数内部做的事情其实就是去查找对象有没有实现__iter__方法或者在老式协议里有没有实现__getitem__方法这个后面细说。用生活化的类比来解释可迭代对象就像一叠放在桌上的扑克牌你知道里面有 54 张牌可以一张一张翻看但牌本身不会告诉你“你翻到第几张了”。每次你从头开始翻都会从第一张开始。2.2 迭代器是记住了进度的一条通道迭代器Iterator是实现了迭代协议、能记住遍历位置的对象。它有两个核心方法__iter__()返回迭代器自身。这个方法让迭代器本身也是可迭代对象所以它也能被 for 循环。__next__()返回下一个值。没有更多值时抛出StopIteration异常。继续用扑克牌做类比迭代器不是那叠牌而是一只插在牌堆里的手指。它不仅知道下一张牌是什么还知道你翻到哪里了。每调用一次next()手指就往下移一张牌翻完了它就喊一声 StopIteration。关键区别就在这里可迭代对象是“静态的资源”迭代器是“带状态的通道”。2.3 迭代协议__iter__与__next__的契约Python 的迭代协议本质上就是一份契约一个对象如果实现了__iter__()它就是可迭代对象一个对象如果同时实现了__iter__()和__next__()它就是迭代器用iter(obj)调用时Python 自动去调用obj.__iter__()拿到一个迭代器用next(it)调用时Python 自动去调用it.__next__()拿到下一个值或者捕获StopIteration。用代码直观地展示这两层关系# 列表是可迭代对象但不是迭代器 my_list [1, 2, 3] print(hasattr(my_list, __iter__)) # True print(hasattr(my_list, __next__)) # False列表没有 __next__ # 对列表调用 iter()得到一个真正的迭代器 list_iter iter(my_list) print(hasattr(list_iter, __iter__)) # True print(hasattr(list_iter, __next__)) # True # 手动驱动迭代器 print(next(list_iter)) # 1 print(next(list_iter)) # 2 print(next(list_iter)) # 3 print(next(list_iter)) # StopIteration这就是为什么list本身不是迭代器而iter(list)的返回值是迭代器。这个区别是理解一切后续内容的基石务必先消化掉。3. 把 for 循环的遮羞布掀开它到底是怎么工作的很多教程告诉你“for 循环就是语法糖”但没说清楚糖衣里面裹的是什么。这一节我直接把 for 循环的底裤扒干净。3.1 手动模拟 for 循环的执行过程当你写下for item in [1, 2, 3]: print(item)Python 实际执行的是下面这个过程_iter iter([1, 2, 3]) # 第1步拿到迭代器 while True: try: item next(_iter) # 第2步逐个取下一个值 except StopIteration: # 第3步没有值了跳出循环 break print(item) # 第4步执行循环体也就是说for 循环的完整语义等价于“获取迭代器 循环调 next 捕获 StopIteration 异常退出”三件事。你平时写的每一行 for 循环底层都在走这条链路。理解这个机制之后下面几个结论就水到渠成了for 循环并不要求对象是列表只要它是可迭代对象就行for 循环之所以能遍历文件对象因为文件对象实现了__iter__()和__next__()它本身就是迭代器字典 for 循环默认遍历的是 key因为字典的迭代器按 key 产出。3.2iter()函数的两副面孔正经协议和旧式协议iter()函数其实有两种用法很多人只见过第一种第一种iter(iterable)把一个可迭代对象变成迭代器。这是最常见的形式背后的查找顺序是先找__iter__方法调它拿到迭代器如果对象没有__iter__再找__getitem__这种旧式协议。第二种iter(callable, sentinel)反复调用一个可调用对象直到返回哨兵值。这个用法相当冷门但实际效率极高。最常见的例子是逐块读取文件with open(data.bin, rb) as f: for chunk in iter(lambda: f.read(4096), b): process(chunk)iter(lambda: f.read(4096), b)的意思是每次迭代都调用一次f.read(4096)如果返回值是b空字节串就停止。这样写比手写while True加 break 简洁得多而且逻辑一眼清晰。面试时如果你能顺手甩出这个写法观感会很不一样。3.3 为什么列表可以反复 for迭代器只能走一次这个问题是我让学生做的第一个小实验效果非常震撼my_list [1, 2, 3] for i in my_list: print(i, end ) # 1 2 3 for i in my_list: print(i, end ) # 1 2 3列表可以反复遍历my_iter iter([1, 2, 3]) for i in my_iter: print(i, end ) # 1 2 3 for i in my_iter: print(i, end ) # 什么都不打印迭代器已经耗尽了为什么因为列表只是可迭代对象每次 for 循环都会调用iter(my_list)生成一个全新的迭代器所以可以无限次从头遍历。而迭代器本身带有状态第一次 for 循环已经把它推到了StopIteration第二次 for 时哪怕再调iter(my_iter)返回的还是它自己指针已经在末尾自然什么都取不出来。这个特性用得好是省内存的利器用不好就是“那个折磨我一小时的 bug”。4. 生成器写迭代器最偷懒的方式没有之一每次我讲完手写__iter__和__next__都会看到有人表情痛苦。别急Python 早就给你准备了捷径——生成器。它本质就是一个用yield关键字写出来的迭代器但写起来像普通函数行为却完全符合迭代器协议。4.1 yield 生成器的行为分析看一个最经典的斐波那契生成器def fibonacci(): a, b 0, 1 while True: yield a a, b b, a b fib fibonacci() print(next(fib)) # 0 print(next(fib)) # 1 print(next(fib)) # 1 print(next(fib)) # 2fibonacci()是个函数但函数体内只要出现了yield它就不再是普通函数而是生成器函数。调用fibonacci()不会执行函数体任何一行代码只是返回一个生成器对象。每次next()时函数执行到yield a暂停把a的值交出去再次next()时从暂停的地方继续往下走。这就是生成器和普通函数最大的区别普通函数是一次性执行完返回结果生成器是分段执行、随时暂停、随时恢复。你不需要手动维护状态变量函数内的局部变量在暂停期间都会被保留这是生成器对开发者最友好的一点。4.2 生成器表达式列表推导式的省内存版Python 还有一种更紧凑的生成器写法——生成器表达式。长得跟列表推导式几乎一样只是把方括号换成圆括号# 列表推导式一次性生成全部数据占用内存 squares_list [x * x for x in range(10_000_000)] # 约 320MB 内存 # 生成器表达式惰性求值逐个生成几乎不占内存 squares_gen (x * x for x in range(10_000_000)) # 几乎 0 内存注意一个小坑生成器表达式是一次性的遍历完就没了。如果你需要反复用同一批数据要么转成列表要么重新创建生成器。4.3 生成器函数还是生成器表达式我的选择标准逻辑简单比如(x * x for x in data)用生成器表达式一行搞定逻辑复杂比如需要多步处理、异常处理、状态保持用生成器函数可读性更强需要反复遍历不要用生成器老老实实返回列表或元组。4.4 协程与 yield from生成器的高级姿势yield不仅能产出值还能接收值这就是协程的基础。看这个例子def echo(): while True: received yield print(f收到: {received}) e echo() next(e) # 启动生成器走到 yield 处暂停 e.send(hello) # 收到: hello e.send(world) # 收到: worldsend()方法可以把值发送进生成器内部yield表达式的返回值就是send()传来的值。这个机制是现代异步编程的基石之一。yield from则是生成器之间的“委托”。它可以让你在一个生成器里直接迭代另一个可迭代对象或生成器把子生成器的产出“透传”给调用方def chain(*iterables): for it in iterables: yield from it for x in chain([1, 2], ab, (3, 4)): print(x, end ) # 1 2 a b 3 4写自定义拼接逻辑时yield from比手写嵌套循环干净太多。5. 自定义迭代器实战手写一个可回放的 CSV 行读取器理论知识讲了半天不动手永远记不牢。这一节我们从零实现一个自定义迭代器场景选得非常实际读取 CSV 文件但要跳过无效的注释行以 # 开头并且能记录当前行号。这两个需求用现成的csv模块不一定好处理但自定义迭代器可以轻松覆盖。5.1 需求与设计输入CSV 文件的路径行为每次迭代返回一个(行号, 字段列表)元组规则以#开头的行视为注释跳过不返回附加统计有效数据行数。5.2 第一版实现用生成器函数先看用生成器怎么实现代码少思路直观def read_csv_with_comments(file_path): line_number 0 valid_count 0 with open(file_path, r, encodingutf-8) as f: for raw_line in f: line_number 1 stripped raw_line.strip() if not stripped or stripped.startswith(#): continue fields [field.strip() for field in stripped.split(,)] valid_count 1 yield line_number, fields print(f总行数: {line_number}, 有效数据行: {valid_count})用的时候data read_csv_with_comments(data.csv) for line_no, fields in data: print(line_no, fields)这已经完整体现了“惰性 状态保持”的威力。line_number和valid_count是生成器函数的局部变量每次yield暂停时它们都被保存在生成器内部不需要额外的类来管理状态。5.3 第二版实现用类实现完整迭代器协议如果需求还要增加更多能力比如需要单独的方法来获取统计信息、需要支持反复从头读取那么用类实现更合适class CSVLineReader: def __init__(self, file_path): self.file_path file_path self.total_lines 0 self.valid_lines 0 self._file None def __iter__(self): # 每次 __iter__ 都重新打开文件保证可以反复迭代 if self._file: self._file.close() self._file open(self.file_path, r, encodingutf-8) self.total_lines 0 self.valid_lines 0 return self def __next__(self): if not self._file: raise StopIteration for raw_line in self._file: self.total_lines 1 stripped raw_line.strip() if not stripped or stripped.startswith(#): continue self.valid_lines 1 fields [field.strip() for field in stripped.split(,)] return self.total_lines, fields # 文件读完关闭并结束迭代 self._file.close() self._file None raise StopIteration def stats(self): return self.total_lines, self.valid_lines注意__iter__里我做了两件事关闭上一次可能未读完的文件、重置计数器。这样同一个对象可以被 for 循环多次每次都是全新遍历。这不只是为了演示而写是实际项目里踩过坑之后总结出来的经验——如果一个迭代器还打算被重复使用一定不要在__init__里打开资源要在__iter__里处理。5.4 协议实现的细节校验写完之后建议做两件事第一手动测试连续 next 的行为确认最后一行的StopIteration正确抛出reader CSVLineReader(data.csv) it iter(reader) print(next(it)) # 第一行有效数据 # ... 一路 next 到结尾最终抛 StopIteration第二确认对象能直接用于 for 循环reader CSVLineReader(data.csv) for line_no, fields in reader: print(line_no, fields) print(reader.stats())这个类实现中最容易出的 bug 是__next__里没有正确处理文件读完的边界导致抛出的不是StopIteration而是ValueError: I/O operation on closed file。写的时候可以刻意留一个坑让读者自己想清楚为什么__next__返回后要立刻判断文件状态。这种边界条件恰恰是面试官最爱挖的地方。6. 那些年我踩过的迭代器坑四个高频翻车现场迭代器机制本身不复杂但它在实际工程里的表现经常让人抓狂。下面四个坑每个都是我或我带的同事在生产环境里真实遇到过的单独列出来给你避雷。6.1 “我的列表怎么被消费掉了”——把迭代器误当列表某次数据清洗的脚本同事把csv.reader(f)的返回值直接存成变量第一段代码遍历完后第二段代码再用这个变量发现是空的。他百思不得其解最后打印类型才发现问题csv.reader返回的不是列表而是迭代器只能消费一次。排查方法不确定一个对象是不是迭代器时用两个快捷方式之一from collections.abc import Iterator, Iterable print(isinstance(data, Iterator)) # True 说明是一次性的 print(isinstance(data, Iterable)) # True 仅说明可迭代可能还保留数据应对手段如果确实需要反复使用在第一次迭代时就转成列表或者重新构建可迭代对象。6.2 在同一个迭代器上嵌套循环死循环之外还有更隐蔽的新手容易犯的错误是对同一个迭代器嵌套 for。比如想对生成器做“两两比较”data (i for i in range(5)) for a in data: for b in data: # 内层循环直接耗尽 data print(a, b)结果只会输出(0, 1) (0, 2) ...里很少几项然后 data 就被内层循环消费光了外层随之退出整个输出残缺不全。这不是死循环但比死循环更隐蔽因为它能跑只是结果不对。应对手段需要二次遍历的数据先缓存成列表data list(i for i in range(5)) for a in data: for b in data: print(a, b)6.3next()的默认值陷阱不要指望它自动返回 Nonenext(iterator, default)的第二参数是默认值很多人以为传了默认值就不会抛异常。是这样没错但很多人不知道只有迭代器真正耗尽时才返回 default迭代过程中如果抛出其他异常default 是救不了你的。def bad_generator(): yield 1 raise ValueError(boom) it bad_generator() print(next(it, fallback)) # 1 print(next(it, fallback)) # ValueError: boomdefault 不生效这个坑在写健壮性代码时尤其重要。如果你预计迭代过程中数据可能损坏别指望 default 兜底要用 try/except 包住next()调用。6.4 自定义类既没__iter__也没__next__但有__getitem__也能被迭代这是一个冷知识老式迭代协议里只要类实现了__getitem__并且索引越界时抛IndexError它也能被 for 循环遍历。class OldStyleIterable: def __init__(self, items): self.items items def __getitem__(self, index): if index len(self.items): raise IndexError return self.items[index] for x in OldStyleIterable([a, b, c]): print(x) # a b c这个机制保证了兼容 Python 2 时代的老代码。日常写新代码我强烈建议不要依赖它用标准的__iter__协议但看到别人代码里只有__getitem__却能 for 时别觉得是魔法知道来路就行。7. 实战进阶用 itertools 把迭代器玩出花来itertools是 Python 标准库里专门为迭代器设计的工具箱里面的函数全部返回迭代器天然惰性、天然省内存。多数教程只讲chain和count但实际工作中还有很多高频且好用的工具我按使用频率给你列一下。7.1 无限迭代器与限流搭配count、cycle、repeat不知道你有没有想过生成无限序列itertools.count(start0, step1)就是这样一个无限计数器import itertools for i in itertools.count(10, 2): if i 20: break print(i) # 10 12 14 16 18 20注意无限迭代器必须配合 break 条件或takewhile使用否则死循环from itertools import count, takewhile evens takewhile(lambda x: x 20, count(10, 2)) print(list(evens)) # [10, 12, 14, 16, 18, 20]cycle可以让一个序列无限循环适合做轮流分配任务from itertools import cycle nodes cycle([node1, node2, node3]) for i in range(10): print(next(nodes)) # node1 node2 node3 node1 ...7.2 组合迭代器chain、zip_longest、tee、islicechain把多个可迭代对象串成一个我经常用来组装多个配置源from itertools import chain defaults {host: localhost, port: 8080} env_config {port: 9090} merged dict(chain(defaults.items(), env_config.items())) print(merged) # {host: localhost, port: 9090}zip_longest处理长度不一致的多个迭代器用默认值补齐缺失项from itertools import zip_longest names [Alice, Bob] scores [95, 87, 76] for name, score in zip_longest(names, scores, fillvalueN/A): print(name, score) # Alice 95 # Bob 87 # N/A 76tee把一个迭代器“复制”成多个独立副本注意它是用内存换取多次迭代能力复制越多占内存越多from itertools import tee source (x * x for x in range(5)) a, b tee(source) # 生成两个可以独立遍历的迭代器 print(list(a)) # [0, 1, 4, 9, 16] print(list(b)) # [0, 1, 4, 9, 16]islice对迭代器做切片注意它不接受负数索引但可以用None表示起点from itertools import islice data iter(range(100)) head list(islice(data, 5)) # 取前5个 mid list(islice(data, 10, 20)) # 从当前状态再取第10到第19个 print(head) # [0, 1, 2, 3, 4] print(mid) # [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]7.3 实战用 itertools 处理实时日志流把上面几个工具组合起来可以实现一个非常小巧的日志监控脚本持续读取日志文件新增行过滤出 ERROR每 10 条打一个摘要。import itertools import time def follow(file_path): with open(file_path, r, encodingutf-8) as f: f.seek(0, 2) # 跳到文件末尾 while True: line f.readline() if not line: time.sleep(0.1) # 没有新行等待 continue yield line def extract_errors(lines): for line in lines: if ERROR in line: yield line.strip() def batch(iterable, size): return iter(lambda: list(itertools.islice(iterable, size)), []) for batch_lines in batch(extract_errors(follow(app.log)), 10): print(f--- 发现 {len(batch_lines)} 条错误 ---) for line in batch_lines: print(line)follow函数模拟 Unix 的tail -f就是一个无限生成器extract_errors负责过滤batch用之前讲的iter(callable, sentinel)把输出按 10 条分组成批次。整个脚本没有一行复杂的循环嵌套读起来像流水线一样清晰。这就是迭代器思维带来的代码组织方式的改变。8. 性能与取舍什么时候该用迭代器什么时候别硬用说了迭代器这么多好处我必须客观点不是所有场景都适合迭代器。盲目追求“惰性”反而会让代码变复杂。我给自己定了几条实践准则供你参考。8.1 迭代器占内存小的本质并不是没有占用很多人理解成“迭代器不占用内存”这是错的。生成器不一次性把全部数据放进内存但它每次产出数据时数据本身占用的内存还是存在的只是被及时释放了。文件迭代器读一行处理一行那行字符串在内存中的生命周期极短。本质是空间换时间的对立面——它用算力逐次执行换内存不批量持有。8.2 小数据量别硬上迭代器如果你要操作的数据只有几十个元素用列表推导式生成完整列表代码更简洁、调试更方便性能和内存几乎没有差别。这时候硬生生搞一个多层生成器纯粹是给自己找麻烦。我的选择标准数据量明确很小比如配置文件里的几行、固定枚举值用列表/元组数据量可能很大或者不确定日志、用户上传、分页结果用迭代器需要一个接一个处理流式数据网络流、实时数据必须用迭代器/生成器需要随机访问、反复遍历、求长度、做切片优先转成列表。8.3 迭代器的调试成本迭代器有一个非常讨厌的缺点你无法直接查看它的内部内容。print(list(my_iter))虽然能看但看一次就把迭代器消费光了调试完原逻辑就废了。我的调试技巧是用itertools.tee复制一份来打印原迭代器不受影响from itertools import tee debug_iter, real_iter tee(original_iter) print(list(debug_iter)) # 打印第一份方便检查 for item in real_iter: # 用第二份继续正常业务 process(item)8.4 嵌套生成器的可读性问题生成器一层套一层很容易写出“一眼看不懂在干嘛”的代码。我见过有人写五层嵌套生成器来处理数据流优雅是优雅维护的人差点崩溃。处理办法是给每层生成器起一个“动词性”的函数名让它像流水线工位一样清晰raw_lines - normalize_lines - filter_valid - extract_fields。迭代器的威力不是靠嵌套深度体现的是靠清晰的边界和职责划分体现的。最后分享一个我实际操作中的小习惯每学一个新的容器或工具我都会先用hasattr和isinstance把它在迭代器协议中的位置查一遍比如“range 对象是不是迭代器”“dict 的 values 视图能不能重复遍历”。查完随手写下结论久而久之迭代器的概念就内化成直觉了。遇到诡异 bug 时先问自己一句“这里的数据到底是可迭代对象还是迭代器”往往一句话就能定位问题根源。