先抛一个问题你写了这么久的 Python有没有想过for x in obj:这行代码到底在背后干了什么我自己第一次认真琢磨这件事是在排查一个诡异的线上 bug 时——一个看起来人畜无害的 for 循环竟然把一个 16G 内存的机器跑到了 OOM。后来才发现问题出在一个自定义类把__iter__写成了返回列表而列表里又套着列表局面完全失控。从那天起我就意识到迭代协议不是文档里那种“读懂就行”的概念。它是 Python 里最底层、最高频、也最容易被忽视的机制之一。你每天在用的列表推导式、for 循环、解包语法a, b ...甚至dict.get背后那套查找逻辑全都在跟迭代协议打交道。再往后看你接触 asyncio、async for、异步生成器时如果没啃透同步迭代协议你会非常痛苦——因为异步迭代就是照着同步迭代的骨架长出来的。这篇文章我会把迭代协议完整拆开从__iter__/__next__/__getitem__三个核心方法讲起到手写迭代器、生成器、yield再到 async for 和异步生成器最后附上我在实践中踩过的一堆坑。不管是刚入门 Python 的新手还是已经在写协程但总觉得“差一口气”的进阶选手这篇文章都能帮你把那层窗户纸捅破。1. 迭代协议是解开 for 循环和 async 的一把钥匙1.1 先搞清楚for 循环并不是魔法很多人学 Python 的第一课就是for i in range(10)然后老师告诉你“这是遍历”。但“遍历”这个词太模糊了。你真的知道 range(10) 在底层返回了什么吗知道 for 循环是“一次一次调 next()”的吗实际上Python 的 for 循环可以被完全展开成下面这套底层逻辑# 你写的 for x in obj: print(x) # 解释器内心实际上是这样跑的 it iter(obj) # 1. 先拿到迭代器 while True: try: x next(it) # 2. 反复调用 next() except StopIteration: break # 3. 迭代器“没货了”主动退出 print(x)这短短三步就是迭代协议的全部骨架。iter(obj)负责把对象变成迭代器next(it)一次拿一个元素拿光了抛StopIterationfor 循环捕获这个异常后静默退出。理解这一点你就能解释很多“怪现象”为什么文件对象可以一行一行读却没法把这个文件对象在列表推导式里重复用两次为什么生成器只能迭代一次为什么list(map(str, numbers))里的 map 对象你用完了之后再转 list 就是空的这些问题的答案全都在“迭代器是一次性消耗品”这条铁律里。1.2 异步编程为什么必须依赖迭代协议说完了 for 循环我们再来看异步编程。async for出现了而且长得和 for 循环几乎一模一样。为什么会这样因为在 Python 的设计哲学里不管同步还是异步“从某个数据源逐个获取元素”这个需求是通用的。同步世界是用__next__直接拿下一个值拿不到就抛StopIteration。异步世界则需要考虑一个关键问题拿下一个值这个操作本身可能耗时比如从网络 socket 里读数据、从数据库游标里取记录、从消息队列里拉任务。在这些场景里你不可能用阻塞的方式去“等下一个值”那会把整个事件循环卡死。于是 Python 在 3.5 引入了异步迭代协议__aiter__和__anext__。它和同步协议几乎一一对应唯一区别是__anext__是一个协程返回的是 awaitable 对象。异步迭代耗时的操作时事件循环可以在等待 IO 的同时去处理别的任务这才是 asyncio 高性能的来源。所以我的建议是不要直接去硬啃 asyncio 那一大堆概念先回到迭代协议本身。你把同步迭代协议吃透了异步迭代就是换了个马甲的同一匹马。2. 底层原理拆解iter、next和那条古老的回退协议2.1 iter() 到底做了什么先找iter再退到getitemiter(x)在 Python 里的查找顺序是这样的先看 x 有没有__iter__方法有就调用拿返回值作为迭代器。如果没有__iter__再看有没有__getitem__方法有就返回一个特殊的迭代器这个迭代器会按下标 0、1、2、3……一直取下去。两个都没有抛TypeError: xxx object is not iterable。第二条经常被忽略。这是 Python 为了兼容老代码保留的“旧式迭代协议”但它在今天依然有效。比如 dict 在早期版本里没有__iter__直接暴露迭代逻辑时就是用__getitem__来支持按键迭代的。我来给你演示一个“没有__iter__也能迭代”的例子class OldStyle: def __init__(self, items): self._items items def __getitem__(self, index): return self._items[index] obj OldStyle([a, b, c]) for item in obj: print(item) # 输出 a b c这段代码能跑不是因为 for 循环认识OldStyle而是因为iter(obj)发现它没有__iter__就退而求其次用__getitem__(0)、__getitem__(1)……一直取到IndexError才停止。没错对这类对象来说IndexError就是它们的StopIteration。你可以玩玩一个更有意思的无限序列class Infinity: def __getitem__(self, index): return index * index for i in Infinity(): print(i) if i 100: break因为__getitem__永远不会抛 IndexError这个对象理论上可以无限迭代下去。是不是想起了itertools.count本质上都是同一个思路。2.2 可迭代对象和迭代器别再混为一谈这是面试高频题也是很多人写代码翻车的重灾区。这两者的区别用一句话就能说清可迭代对象Iterable实现了__iter__调用iter(obj)能拿到迭代器。比如 list、tuple、dict、set、str。迭代器Iterator实现了__iter__和__next__next(it)能一次取一个值取完抛 StopIteration。典型的比如文件对象、生成器对象。关键区别在于list 是可迭代对象但不是迭代器。你写next([1, 2, 3])会直接报 TypeError因为 list 根本没有__next__方法。那为什么迭代器协议要求迭代器也必须实现__iter__因为 for 循环拿到迭代器之后还会再对它调一次iter()。为了兼容这个行为迭代器最好把__iter__返回自己。这也是为什么 for 循环既能作用在 list 上也能直接作用在生成器上——生成器的__iter__就是返回自己。from collections.abc import Iterator, Iterable print(isinstance([], Iterable)) # True print(isinstance([], Iterator)) # False print(isinstance((x for x in []), Iterator)) # True2.3 StopIteration迭代结束的标准信号StopIteration不仅是个普通异常它其实是整个迭代协议里的“终止信号”。你写生成器时的return本质就是抛一个 StopIterationfor 循环捕获到它才肯退出。这里有个很经典的坑在生成器里 return 一个值。这个值不是给 for 循环用的而是藏在 StopIteration 异常对象上。你要真想拿到它得手动捕获def gen(): yield 1 return done g gen() print(next(g)) # 1 try: next(g) except StopIteration as e: print(e.value) # done平时你写 for 循环这个 value 会被静默丢弃。但如果你在做协程、状态机调度这类底层封装这个机制就很有用。3. 核心实践手写一个可迭代对象3.1 一个完整的迭代器类长什么样理论说再多不写代码都是纸老虎。我们来手写一个完整的、符合现代人习惯的迭代器类。功能很朴素模拟一个固定步长的数字序列类似 range 的极简版。class StepRange: def __init__(self, start, stop, step1): self._start start self._stop stop self._step step def __iter__(self): self._current self._start return self def __next__(self): if self._current self._stop: raise StopIteration value self._current self._current self._step return value测试一下sr StepRange(1, 10, 2) for i in sr: print(i) # 1 3 5 7 9 # 重点测试重复迭代 for i in sr: print(i) # 1 3 5 7 9因为 for 循环会先调 iter()重置了 _current为什么第二次 for 循环还能从头开始因为我把__iter__里做了状态重置。这是“可迭代对象”和“一次性迭代器”的分水岭可迭代对象每一次 for 循环都能产生一个全新的迭代器而迭代器自身用完就废。如果你不想自己维护__iter__里的重置逻辑还有一个更简单的做法永远在__iter__里返回一个全新的迭代器对象。class StepRangeV2: def __init__(self, start, stop, step1): self._start start self._stop stop self._step step def __iter__(self): return self._Generator(self._start, self._stop, self._step) class _Generator: def __init__(self, start, stop, step): self._current start self._stop stop self._step step def __iter__(self): return self def __next__(self): if self._current self._stop: raise StopIteration value self._current self._current self._step return value两种写法风格不同前者省内存后者更符合“每次迭代独立”的语义。实战里我个人更常用后者因为它天然规避了“同一时刻多个迭代器互相踩状态”的隐性 bug。3.2 为什么要自己写迭代器惰性求值的真香时刻你可能会问我直接用 list 不香吗为什么要费劲写迭代器核心答案就三个字惰性求值。迭代器是“边取边算”的。它天然适合处理两种情况数据量太大或者计算代价太高。举个例子你要遍历一个 100 亿个整数的序列每个值是前一个的平方模一个大素数。你不可能先造出这个 100 亿长度的 list内存直接爆掉。但用迭代器写内存占用是常量级。class ModularSquare: def __init__(self, n, seed1): self._n n self._seed seed def __iter__(self): return self def __next__(self): value self._seed self._seed (self._seed * self._seed) % self._n return value ms ModularSquare(1000000007) for i, v in enumerate(ms): print(v) if i 1000: break另一个更现实的好处是读取大数据文件时for line in f:能够一行一行处理内存占用始终很低。如果你写成for line in f.readlines():几 GB 的文件会直接把内存吃干。这个区别的本质就是迭代协议在背后做的事。3.3 迭代器协议和容器协议怎么配合一个合格的自定义容器类通常不只是能迭代还应该有len()、in、[]下标访问这些基础能力。这里就得用到容器协议__len__和__getitem__。class MyCollection: def __init__(self, data): self._data list(data) def __iter__(self): return iter(self._data) def __len__(self): return len(self._data) def __getitem__(self, index): return self._data[index]有了__len__len(obj)就能用有了__getitem__obj[0]能用而且——你看回第 2.1 节——就算你不写__iter__for 循环老协议也能让它遍历。但这里强烈建议你还是把__iter__写了因为老协议的性能不行它要从下标 0 开始硬取还只能支持顺序访问语义也不完整。注意bool(obj)的判断依赖__len__或__bool__。如果你的类只实现了__iter__没有实现__len__在 if 里判断永远为 True。这个坑非常隐蔽我第一次写自定义集合时就在这里翻车。4. 生成器迭代协议在 Python 里的最终形态4.1 yield 背后其实就是一个迭代器前面手写类的方式你已经见识了迭代协议的完整面貌。但实际工作里90% 的情况下你不会这么写——你会直接写生成器函数。因为生成器就是“自动帮你实现了完整迭代器协议”的语法糖。def step_range(start, stop, step1): current start while current stop: yield current current step这个函数没有 return也没有类定义但它可以被 for 循环直接迭代。原因就是只要函数体里出现 yield调用这个函数就不再执行函数体而是返回一个生成器对象。生成器对象内部已经内置了__iter__和__next__你 next 一次函数体就跑到下一个 yield把 yield 右边的值交给你然后挂起。yield 和 return 的本质区别在于return 是退出函数yield 是挂起函数。挂起意味着局部变量、执行位置全部保留下次 next 再接着往下执行。这个“挂起/恢复”的能力正是协程的雏形。4.2 生成器表达式和 itertools给你的迭代器加点料生成器表达式是列表推导式的“懒兄弟”# 列表推导式立即把所有元素算完存在内存里 squares [x * x for x in range(1000000)] # 生成器表达式惰性求值用的时候才算 squares_gen (x * x for x in range(1000000))第二个写法的内存占用是常量级的性能差异在百万级数据量上非常明显。但注意生成器表达式是一次性消耗品你不能像 list 那样反复遍历也不能用 len()。如果你觉得手动写生成器逻辑繁琐标准库itertools就是你的瑞士军刀。count造无限计数、cycle无限循环、groupby分组、chain拼接、islice切片——这些工具吃透之后很多原本要写几十行的数据处理逻辑一行就能搞定。from itertools import islice, count # 取 1 到 100 之间的偶数平方的前 10 个 evens (x * x for x in count(2, 2)) result list(islice(evens, 10)) print(result)islice解决了一个大问题普通的生成器不支持切片。有了它你可以对无限序列做裁剪。4.3 生成器的 send、throw 与协程根源yield 不仅能“吐”值还能“接”值。这就是send()做的事。用 send 向生成器内部传数据可以实现双向通信这也是协程的理论基础。def accumulator(): total 0 while True: value yield total # yield 会把 send 进来的值赋给 value if value is None: continue total value acc accumulator() print(next(acc)) # 0先预热到第一个 yield print(acc.send(10)) # 10 print(acc.send(5)) # 15注意两个细节第一创建生成器后必须先调用一次 next() 或者 send(None)让它执行到第一个 yield第二不能在生成器还没启动时直接 send 非 None 值否则会抛 TypeError。throw()和close()是另外两个配套能力前者往生成器内部抛异常后者强制关闭生成器并抛 GeneratorExit。这些机制后面全部被 async 协程所复用。Python 官方甚至在最早期用生成器来实现协程你如果去读旧代码里基于 yield from 的协程实现会发现它就是 async/await 的前身。5. 迭代协议如何成为异步编程基石5.1 事件循环和协程先建立直觉在谈async for之前必须先建立事件循环的直觉。事件循环有点像餐厅里的领位员你点完菜发起了 IO 请求他让你先坐着挂起当前协程等菜做好了IO 完成再叫你恢复协程继续执行。在这个模型里一个协程就是“可以被挂起和恢复的函数”而让它挂起的关键就是 await 一个耗时操作。import asyncio async def task(name, delay): print(f{name} 开始) await asyncio.sleep(delay) print(f{name} 结束) async def main(): await asyncio.gather( task(A, 1), task(B, 2), task(C, 1), ) asyncio.run(main())三个任务总共耗时约 2 秒而非 4 秒靠的就是事件循环在 await 挂起期间调度其他协程。这条基线和迭代协议有什么关系往下看。5.2aiter/anext异步迭代协议的真身当你要异步地“逐个拿下一个值”时同步迭代协议就撑不住了。因为__next__是同步函数它拿下一个值的时候如果遇到网络 IO只能阻塞等待。阻塞在事件循环线程里等于把所有协程全堵死。Python 因此定义了异步迭代协议class AsyncCounter: def __init__(self, limit): self._limit limit self._current 0 def __aiter__(self): return self async def __anext__(self): if self._current self._limit: raise StopAsyncIteration await asyncio.sleep(0.1) # 模拟异步耗时操作 self._current 1 return self._current使用方式async def main(): async for i in AsyncCounter(5): print(i) asyncio.run(main())async for的底层逻辑如下it aiter(obj) # Python 3.10 内置函数等价于 obj.__aiter__() while True: try: x await anext(it) # 3.10 内置函数等价于 await obj.__anext__() except StopAsyncIteration: break print(x)看到了吧和同步 for 循环的结构完全一致只是把 next() 换成了 await anext()。同步协议里 StopIteration 对应异步协议里 StopAsyncIteration同步迭代器返回自己对应异步迭代器也返回自己。这套一一对应的映射关系就是你从同步理解异步的桥。5.3 异步生成器与实际落地场景和同步生成器一样异步迭代器也别手写类了直接用异步生成器更香。语法上就是async def函数里用yieldasync def read_lines(stream): while True: line await stream.readline() if not line: break yield line这个函数每次 yield 之前都会等一个异步 IO 完成。于是你可以在事件循环里优雅地处理大文件、数据库游标、WebSocket 消息流、Kafka 消费等场景。最好用的标准库例子是asyncio.open_connection之后按行读取网络数据async def echo_client(host, port): reader, writer await asyncio.open_connection(host, port) writer.write(bhello\n) await writer.drain() async for line in reader: print(f收到: {line.decode().strip()})是不是感觉整个代码像同步代码一样平顺这就是 async for 对心智负担的极大缓解。你不需要再手动维护一个 while 循环里 await 读数据、再判断是否结束的样板代码了。5.4 异步迭代器和同步迭代器混用的坑这里我在生产环境踩过一个非常典型的坑在异步代码里直接用了同步迭代器去遍历一个耗时的操作结果整个事件循环被卡死。# 错误示范 async def bad_worker(): for url in urls: data await fetch(url) # 没问题是异步的 for item in data: # data 是普通 list同步迭代很快没问题 process_slowly(item) # 但这里的 process_slowly 是阻塞的卡死循环真正的坑是同步生成器函数不能直接变成异步迭代器你没法在 async for 里用同步生成器。如果我们在 async for 的循环体里有阻塞操作整个事件循环没人能让出 CPU。所以异步代码里的原则是循环体内部尽量不要出现阻塞式的同步调用真要跑 CPU 密集任务丢到asyncio.to_thread里去。反过来同步世界也不能直接用异步迭代器。你不能写for x in AsyncCounter(...)编译直接报错。跨协议使用必须自己包装。6. 常见问题与排查技巧实录6.1 高频报错和逻辑坑速查我整理了工作中最常遇到的迭代相关问题和解决方案做成一张速查表现象根因解决办法object is not iterable类没实现__iter__也没有__getitem__补上迭代协议方法同一个生成器第二次 for 循环为空生成器是一次性迭代器重新调用生成器函数或用 itertools.tee 分支next()取不到值却没捕获 StopIteration手动调用 next 未处理结束信号用 for 循环代替或包 try/except异步迭代一直卡住不退出__anext__忘记抛StopAsyncIteration检查结束条件确保最终抛 StopAsyncIteration自定义类在 if 里永远是 True只实现__iter__没实现__len__/__bool__按需求实现长度或布尔语义迭代过程中修改 list 抛 RuntimeError容器迭代器和容器本身状态冲突修改前先转成 list或用 list copyasync for拿到的元素不是预期顺序多个协程并发消费同一个异步迭代器用队列显式串行/并行控制6.2 三个亲自踩过的深坑第一个坑是“无限迭代器没设出口”。我写过一段程序用 while True 的生成器生成随机数做蒙特卡洛模拟结果在某次重构中把 break 条件丢了进程直接“假死”。后来每次写有 while True 的生成器我都会在注释里标明退出条件并且给生成器加一个最大迭代次数的保护参数。第二个坑是“生产者消费者模式里消费者忘了 close 生成器”。生成器对象虽然会被垃圾回收但如果它里面有个 finally 块或者资源句柄比如打开的文件、数据库连接你不显式调用 close 就可能导致资源泄漏。教训是凡是生成器内部用了资源的外层一定要用 try/finally 或 contextlib.closing 包一层。第三个坑是“async for 里面混了同步 IO”。这个最危险因为它不容易报错只是让整个程序性能突然滑坡。我排查过线上一个数据导入任务本来 3 分钟跑完某次上线后突然变成 25 分钟最后定位到是有人在新加的 async for 循环体里用了同步 requests.get。换成await asyncio.to_thread或者直接用 aiohttp性能立刻恢复。6.3 两个能救命的调试技巧排查迭代器状态的时候不要光靠 print。先用isinstance(obj, Iterator)确认对象是不是迭代器再手动 next 几步看状态。排查异步迭代器时用asyncio.current_task()打印当前协程栈配合aiodebug或者asyncio.get_event_loop().set_debug(True)能看到哪个协程长期没让出。另一个技巧是用itertools.tee复制迭代器分支。当你要复用某个昂贵的生成器但不想重新计算时tee 可以模拟出多个独立游标。但注意 tee 本身会在内存里缓存未消费的元素无限迭代器千万别乱用。最后补一句实操心得如果你读完这篇只记住一个东西我希望是这条迭代协议的核心不是“遍历”而是“逐个获取 状态保存 惰性计算”。不管是最简单的 list还是最复杂的 async for底层都在做同一件事。我个人后来的习惯是凡是写自定义容器类先问自己要不要支持迭代支持的话优先用生成器而不是手写迭代器类涉及异步 IO 时第一反应就写async for而不是在同步循环里硬塞 await。这套思维换来的不是更炫的代码而是更少的 bug 和更稳的性能。你今晚就可以试试把自己写过的某个返回 list 的函数改成生成器版本感受一下内存和调用方式的差异——这个体验比看十篇文章都管用。