不少人第一次被迭代器绊住是在写出for i in range(10)之后突然看到某个函数返回了一个generator object ...print 出来是内存地址len() 报错想取第二个值也没有下标于是整个人就懵了。我自己当年刚完成 python 安装、照着 python 基础语法一步步敲代码的时候也是在next()和StopIteration这两个东西上卡了整整一个晚上。这篇内容就是想把什么是 python 的迭代器、迭代器到底有什么好处这件事讲透从协议约定、手写实现、内存实测到生成器的落地用法和排查技巧全部按我实际踩过的顺序写下来。适合正在学 python 基础、刚配好 vscode 或 pycharm 的 python 环境、准备上手数据处理和新项目的人看看完你应该能自己判断这段代码该用列表还是该用迭代器。1. 从一次卡壳说起迭代器到底解决了什么问题1.1 可迭代对象、迭代器、生成器三者不是一回事刚入门的人最容易把这三个词混着用实际上它们的边界很清楚。**可迭代对象iterable**指的是实现了__iter__()方法、或者退一步实现了__getitem__()的对象列表、元组、字符串、字典、集合、文件对象都算。**迭代器iterator**是同时实现了__iter__()和__next__()的对象它自己就是自己的迭代器永远返回 self。**生成器generator**是用yield或者生成器表达式造出来的东西它是迭代器的一种便捷写法但反过来不成立迭代器不一定非得是生成器。判断方法很直接用iter()和next()两个内置函数去试iter(x)不报错说明 x 是可迭代对象next(x)不报错说明 x 本身是迭代器。列表可以iter()但直接next()会报 TypeError这个差异就是区分两者的最快手段。lst [1, 2, 3] it iter(lst) print(next(it)) # 1 print(next(it)) # 2 # next(lst) # TypeError: list object is not an iterator为什么 Python 要费劲区分这两层因为可迭代但不可迭代器的设计允许一个容器被反复遍历。列表每次调用iter()都产出一个全新的迭代器你 for 十次就有十个独立游标互不干扰。而迭代器只有一个游标走完就没了。这个区别看起来是细节实际上决定了你后面所有代码能不能重跑。1.2 为什么 for 循环能跑而手写 next() 会报 TypeError很多人的疑问是既然 for 能遍历列表那为什么我把它拆成 next() 就不行答案在于for语句本身就是个语法糖它在背后做了两件事。第一步调用iter(对象)拿到迭代器第二步反复调用next()直到捕获StopIteration异常就安静地停下来。异常不往外抛循环就正常结束。# 这两段代码在语义上等价 for x in [1, 2, 3]: print(x) _it iter([1, 2, 3]) while True: try: x next(_it) except StopIteration: break print(x)看清楚这个展开之后很多报错就自己解释了。next(lst)报 TypeError 是因为列表没实现__next__Python 不知道怎么推进它的游标。而for不会报错是因为它先调了iter()做转换。注意for捕获的是StopIteration不是所有异常。如果你在迭代器内部写了别的异常比如ValueError程序照样会崩不要以为 for 能兜住一切。这里还有一个容易被忽略的点Python 3.7 之后StopIteration如果从生成器函数里逃逸出来会被自动转换成RuntimeError。这条规则是 PEP 479 定的目的就是防止生成器里误用next()导致迭代莫名提前结束。我当年写一个解析器的时候就在生成器里直接next(inner)结果外层循环悄无声息地少跑了一半数据查了很久才反应过来是这个机制在起作用。1.3 迭代器协议的两条硬性约定想自己写迭代器就得守住两条约定。第一条__iter__()必须返回一个迭代器对象通常直接return self。第二条__next__()每次返回一个元素没有元素了就raise StopIteration注意是 raise 不是 return。第二条最容易被写错。有人图省事写成return None结果迭代器永远不会正常结束你用list()去转它程序直接卡死或者把内存吃光。StopIteration是信号而不是数据它的作用就是告诉消费方我这儿没货了。方法必须返回什么常见错误写法__iter__迭代器对象通常 self忘记定义导致对象不可迭代__next__下一个元素用 return None 代替 raise__next__结束态raise StopIteration抛别的异常for 无法识别还有一个隐性约定迭代器一旦耗尽就应该保持耗尽状态。你连调三次next()前两次拿数据第三次开始每次都该抛StopIteration而不是重置再从头来一遍。违反这条代码在有缓存、有重试逻辑的场景下会出现非常诡异的行为比如重试一次突然又拿到一轮数据。2. 亲手写一个迭代器从iter和next开始2.1 最小可用迭代器的完整代码从零写一个倒计时迭代器比看十篇原理文章管用。class 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 n in Countdown(3): print(n) # 输出 3 2 1这段代码里有三个细节值得说。self.current是状态迭代的本质就是一个有状态的游标 每次推进的规则。__iter__返回 self 是标准写法让这个对象既可以被 for 直接用也可以被iter()转换。__next__里先判断边界再返回顺序不能反先减后判会漏掉最后一个值或者多出一个 0。我建议你在本地把这段敲一遍然后故意把raise StopIteration改成return None看看list(Countdown(3))会发生什么。亲眼看到那个死循环或者超长列表你对信号 vs 数据的理解会比看文档牢固得多。2.2 StopIteration 的边界处理与三种典型写法边界写法主要有三种各有适用场景。第一种是上面那种先判断后返回适合有明确终止条件的场景。第二种是哨兵值写法用一个特殊值标记结束适合处理外部数据源。第三种是用尽即抛依赖内部数据结构的自然耗尽。第二种在实际项目里很常见比如从队列里取数据SENTINEL object() class QueueReader: def __init__(self, data): self._data list(data) def __iter__(self): return self def __next__(self): if not self._data: raise StopIteration item self._data.pop(0) if item is SENTINEL: raise StopIteration return item这里用object()做哨兵而不是用None是因为None在业务数据里可能是合法值用它当终止标记会导致数据被提前截断。这个坑我在处理一批 CSV 的时候就中过某些字段就是空值结果解析到一半就停了排查了半小时才发现哨兵用错了。提示哨兵对象一定要用object()创建的唯一实例比较时用is不要用避免自定义__eq__干扰判断。2.3 一次性耗尽的陷阱与可重启迭代器写法最常被投诉的行为就是迭代器只能用一次。这不是 bug是设计。但很多时候我们确实需要一个能反复遍历的自定义对象这时候正确做法是把__iter__写成生成器函数每次调用返回一个新的生成器而不是让类自己当迭代器。class NumberRange: def __init__(self, start, end): self.start start self.end end def __iter__(self): current self.start while current self.end: yield current current 1 r NumberRange(1, 4) print(list(r)) # [1, 2, 3] print(list(r)) # [1, 2, 3] 可以重复遍历对比一下就能看出设计取向的差别。Countdown是迭代器游标挂在实例上遍历完就废NumberRange是可迭代对象游标是__iter__里的局部变量每次遍历都是全新的。判断标准很简单这个东西需要被遍历几次一次就用迭代器可能多次就用可迭代对象。我个人的习惯是对外暴露的、可能被多个模块复用的容器类一律用第二种写法内部流程里一次性的数据流用第一种或者直接上生成器。这个约定让我的代码在被人接手时少了很多为什么第二次遍历是空的的疑问。3. 迭代器的好处内存、惰性、抽象与解耦3.1 内存占用实测48 字节和 8 MB 的差距迭代器最直观的好处是省内存但到底省多少实测一下更有说服力。import sys lst list(range(1_000_000)) it iter(range(1_000_000)) print(sys.getsizeof(lst)) # 8000056 print(sys.getsizeof(it)) # 48差距接近 16 万倍。为什么会这样因为列表是一次性把结果全部算出来并存在内存里CPython 里列表底层是一个指针数组100 万个元素就是 100 万个 8 字节的指针光数组本身就约 8 MB如果再算上每个 int 对象本身的开销大整数约 28 字节实际占用还要往上翻。而range迭代器只保存三个数当前值、终点、步长总共几十个字节。方式对象大小内存增长规律适用场景list(range(n))随 n 线性增长一次全量需要反复访问、需要下标iter(range(n))常数级恒定单次遍历、数据量大生成器表达式常数级恒定流式处理、管道计算自定义迭代器取决于状态字段通常常数级有复杂状态机的场景不过话说回来迭代器不是无脑替代列表。如果数据量小、需要排序、需要随机访问、需要多次遍历列表就是更好的选择。为了省几百字节把自己的代码写得绕来绕去不值当。我一般的原则是数据规模上万并且只走一遍优先考虑迭代器其余情况先用列表等真的遇到内存瓶颈再换。3.2 惰性求值无限序列和用多少算多少迭代器真正无可替代的地方是惰性求值。它不预先算好所有结果而是每要一个才算一个。这个特性让无限序列成为可能——你没法用一个列表装下无穷多个数但可以用迭代器表达它。def natural_numbers(): n 0 while True: yield n n 1 gen natural_numbers() for _ in range(5): print(next(gen)) # 0 1 2 3 4natural_numbers()永远不会自己结束但你只取 5 个它就只算 5 次内存始终是常数。这个模式在读日志、读传感器流、处理持续产生的业务事件时非常有用。惰性还有一层好处是避免无效计算。比如你要在 1000 万条记录里找第一条满足条件的用列表推导是先把 1000 万条全处理一遍用生成器表达式配合next()找到就停。big (x * x for x in range(10_000_000)) result next(x for x in big if x % 7 0 and x 1000) print(result)这里只算到第一个满足条件的元素就停了后面的 999 万次乘法全部省掉。数据量越大、命中越早省得越多。3.3 管道式数据处理让每段代码只干一件事迭代器的第三个好处是解耦。生产者只管生产中间的处理者只负责转换最后的消费者只负责汇总三方谁都不用关心另外两方的实现细节只要有next()这个统一接口就能拼起来。def read_lines(path): with open(path, encodingutf-8) as f: for line in f: yield line.rstrip(\n) def skip_blank(lines): for line in lines: if line.strip(): yield line def parse_kv(lines): for line in lines: if in line: k, v line.split(, 1) yield k.strip(), v.strip() pipeline parse_kv(skip_blank(read_lines(config.txt))) for k, v in pipeline: print(k, v)这段代码有个很妙的地方文件是边读边处理的不会先把整个文件加载进内存。read_lines一次只产出一行skip_blank一次只处理一行parse_kv一次只解析一行整条链路上没有一处堆积全部数据。第二个妙处是每一层都能独立测试skip_blank接受任何可迭代对象你传列表、传生成器都能跑单元测试不用造真文件。这种数据流水线的写法在数据处理和日志分析里特别常见也是我推荐新手尽早养成的习惯。等你处理过几个几百 MB 的日志文件就会明白边读边算和先全读进来再算的差别有多致命。4. 生成器写迭代器的省事方案4.1 yield 的执行流程逐行拆解手写__iter__和__next__两三个方法还行逻辑一复杂就啰嗦。生成器就是为这个场景准备的语法糖函数里出现yield这个函数就不再是普通函数调用它不会执行函数体而是返回一个生成器对象。def demo(): print(开始) yield 1 print(中间) yield 2 print(结束) g demo() # 此刻什么都不打印 print(拿到生成器) print(next(g)) # 打印开始返回 1 print(next(g)) # 打印中间返回 2 print(next(g)) # 打印结束抛 StopIteration执行流程很关键每次next()把函数从上次暂停的地方继续执行一直跑到下一个yield为止然后挂起并交出值。局部变量、执行位置全部被保存下来下次接着走。挂起期间它不占调用栈不消耗 CPU就静静躺在那里。理解了挂起这个概念很多现象就通了。生成器函数体里第一行在你第一次next()之前根本不会执行所以demo()那行只打印拿到生成器。这也是为什么把文件打开操作放进生成器里要小心——你以为是调用时打开实际上是第一次取值时才打开如果迟迟不取值文件句柄也不会被占用这个特性有时候是优点有时候是 bug 来源。4.2 生成器表达式和列表推导什么时候该选谁[x*x for x in data]和(x*x for x in data)只差一对括号行为差很远。前者是列表推导立刻算出全部结果后者是生成器表达式返回一个惰性对象。对比项列表推导[...]生成器表达式(...)求值时机立即全部计算用到才算内存占用与元素数量成正比常数级可否重复遍历可以不可以是否支持 len/下标支持不支持典型用途结果要多次用只走一遍的中间流选择逻辑其实很简单结果需要被遍历第二次或者需要长度、索引就用列表推导只是一次性的中间态就用生成器表达式。我见过有人为了省内存把所有列表推导都改成生成器表达式结果后面代码里要len()一下只能再list()转回来白白绕一圈。另外一个细节是生成器表达式作为函数唯一参数时括号可以省sum(x for x in data)可以写成sum(x for x in data)但sum((x for x in data))多一层括号也不算错只是没必要。这种小习惯不影响性能但影响代码可读性。4.3 yield from、send 与 close 的实际用法生成器还有几个进阶能力用对了能省不少代码。yield from用来把另一个可迭代对象的值全部转发出来不用写循环def chain_lists(*iterables): for it in iterables: yield from it print(list(chain_lists([1, 2], (3, 4), ab))) # [1, 2, 3, 4, a, b]yield from除了转发值还会自动转发send()和异常行为比手写 for 循环完整。在写递归遍历比如遍历嵌套的目录树、嵌套 JSON时yield from能让代码短一大截。send()是把值送进生成器让生成器从挂起点接收数据继续跑用于协程式写法def accumulator(): total 0 while True: value yield total if value is None: break total value acc accumulator() next(acc) # 先启动走到第一个 yield print(acc.send(10)) # 10 print(acc.send(5)) # 15 acc.close()注意第一次必须先用next()启动生成器否则send()一个非 None 值会报 TypeError。这是刚接触协程写法时最常踩的坑报错信息是cant send non-None value to a just-started generator看到这行就知道忘了启动。close()用来在生成器内部抛GeneratorExit触发finally块适合做资源清理。如果你在生成器里开了文件或连接记得用try/finally包住不然提前close()可能漏掉清理逻辑。5. 实战场景迭代器在大文件、爬虫和数据处理里的落地5.1 大文件读取别再用 readlines 把内存打满处理日志和数据集时最常见的错误是f.readlines()或者f.read().split(\n)。一个 2 GB 的日志文件readlines()会瞬间把内存顶到 2 GB 以上机器直接开始换页速度慢几十倍稍微再大一点就 OOM。正确做法有三种按场景选。第一种是直接迭代文件对象文件对象本身就是可迭代的逐行产出def count_errors(path): n 0 with open(path, encodingutf-8, errorsreplace) as f: for line in f: if ERROR in line: n 1 return n这种方式内存占用几乎恒定一行几百字节到头。第二种是分块读取适合处理定长记录或者二进制文件def read_in_chunks(path, chunk_size1024 * 1024): with open(path, rb) as f: while True: chunk f.read(chunk_size) if not chunk: break yield chunk第三种是带缓冲的流式解析适合单行超长的场景比如某些 JSON Lines 文件里一条记录就有几 MB这时候按行读反而不好得按固定缓冲拼接。我处理过一个每行平均 4 MB 的日志直接用逐行迭代内存峰值到了 8 GB换成 1 MB 分块流式处理之后峰值降到 200 MB 左右。注意f.read()无参数调用会把整个文件读进内存等价于f.read(-1)处理大文件时千万别这么写。5.2 自定义可迭代对象把业务数据源包装成统一接口项目里数据来源五花八门文件、数据库、接口、内存队列如果每个地方都写一套取数逻辑代码会非常散。把它们统一包装成可迭代对象调用方就只需要一个 for 循环。class PaginatedAPI: def __init__(self, fetch_page, page_size100): self.fetch_page fetch_page self.page_size page_size def __iter__(self): page 1 while True: data self.fetch_page(page, self.page_size) if not data: return yield from data if len(data) self.page_size: return page 1这层封装的价值在于调用方拿到了无限数据流的假象只管for item in PaginatedAPI(fetch):翻页逻辑、终止条件、最后一页判断全部藏在迭代器内部。以后接口分页规则改了只改这一个类上层代码一行不动。我踩过的一个坑是终止条件写成if len(data) self.page_size但接口在正好整页时返回空列表表示结束两种规则混用会导致少取一页或者多请求一次。分页终止条件一定要和接口文档确认清楚是在返回为空时停还是在返回数量不足一页时停这两种写法在边界情况下结果完全不同。5.3 迭代器在数据处理里的组合用法数据分析场景里迭代器最典型的组合是读一批、处理一批、写一批避免中间结果堆积。def batched(iterable, size): batch [] for item in iterable: batch.append(item) if len(batch) size: yield batch batch [] if batch: yield batch for batch in batched(read_in_chunks(big.csv), 1000): process(batch)batched这个函数是我写得最多的工具函数之一写数据库时用它做批量插入调接口时用它做批量请求画图时用它控制内存峰值。它本身不依赖具体数据源任何可迭代对象都能塞进来这就是迭代器接口统一带来的好处。需要提醒的是batch []这一行不能省。如果写成batch.clear()然后又 yield 出去你 yield 的列表会被后续清空消费者拿到的全是空列表。生成器里 yield 出来的可变对象一定要在交出控制权之后不再修改它这是个很容易忽视的坑。6. 常见问题与排查技巧实录6.1 迭代器相关报错速查表下面这张表是我整理的高频报错基本都是同一个晚上能撞见三遍的那种。报错信息根本原因解决方式TypeError: list object is not an iterator对可迭代对象直接调 next先iter()转换再 nextTypeError: generator object is not subscriptable对生成器用下标转成 list或改用 itertools.isliceStopIteration冒泡到外层在生成器内直接 next 且没兜住捕获异常并 return或改用 forRuntimeError: generator raised StopIterationPEP 479 自动转换检查生成器内的 next 调用第二次遍历结果为空迭代器已被耗尽改成可迭代对象或重新创建ValueError: too many values to unpack拆包元素数量不匹配用星号收集或先用 list 转换其中第二次遍历为空是新手最常见的问题而且它不报错只是静静给你一个空结果排查起来最费时间。只要看到某个函数的返回值被遍历两次就要警惕是不是迭代器。还有一类是值被吃掉的隐蔽问题。比如你先any(x 5 for x in gen)判断一下再想遍历 gen发现后面的数据全没了因为any已经把整个迭代器走到了尽头。解决办法是把需要多次使用的部分先list()固化或者用itertools.tee复制。不过tee有内存陷阱见下一节。6.2 itertools 里最常用和最易踩坑的几个工具标准库的itertools是迭代器写法的弹药库下面这几个我用得最多islice(iterable, start, stop, step)给任何迭代器加切片能力等价于给流做下标。取前十行直接islice(lines, 10)。chain(a, b)把多个可迭代对象串成一条常用于合并多个数据源。groupby(iterable, key)分组但必须先按同一个 key 排序否则同一个 key 会被切成好几组。count(start, step)无限计数常配合zip给流式数据打编号。repeat(obj, times)重复产出同一个对象做固定次数的重试很顺手。tee(iterable, n)复制成 n 个独立迭代器。tee的坑必须单独说。它内部为每个分支维护一个缓冲队列如果某个分支被消费得特别快、另一个几乎不动那个不动分支待消费的元素会全部缓存在内存里等于是把省内存的优势又还回去了。from itertools import tee a, b tee(range(1_000_000)) list(a) # a 走完了 # 此时 b 的缓冲区里还挂着 100 万个待消费元素 list(b) # 直到这一步才释放所以tee只适合两个分支消费速度接近的场景差距大的话不如老老实实重新生成一遍数据或者把数据落成列表共享。6.3 排查迭代器问题的三条实操经验第一条经验判断一个对象是不是迭代器用iter(x) is x。这个表达式为 True 说明 x 自己就是迭代器为 False 说明它是可迭代对象。比翻源码、看类定义快得多我在交互式环境里基本靠这一手。lst [1, 2, 3] gen (x for x in lst) print(iter(lst) is lst) # False 可迭代对象 print(iter(gen) is gen) # True 迭代器第二条经验调试迭代器不要直接 print用list()或者islice预览。print(gen)只会给你一个内存地址什么信息都没有。想看前几个值就用list(islice(gen, 5))注意这一步会消耗掉前五个元素调试完记得重新构造。如果需要在调试中反复查看就在构造点临时改成列表推导调完再换回来。第三条经验给所有生成器函数加类型注解。def read_lines(path: str) - Iterator[str]:这种写法不只是好看它能让你和接手代码的人一眼看出这个函数返回的是一次性流还是可复用容器。类型标注用Iterator还是Iterable本身就是一份文档。配合 mypy 之类的静态检查还能提前发现对迭代器用了 len这类错误。再说一个日常习惯。我现在写涉及数据流的函数时会在 docstring 里明确写一句这个函数返回迭代器只能遍历一次就这一行备注帮团队省下过不少沟通成本。迭代器的所有坑本质都来自一次性这个特性只要把这个信息传递出去大部分问题在写代码阶段就能避开。