你在写异步代码的时候碰到过TypeError: xxx object is not an async iterator这个报错吗我第一次撞上这个错时第一反应是去查__aiter__折腾半天才发现真正的问题出在缺了__anext__。异步迭代器协议是一体两面的__aiter__负责把迭代器交出来__anext__负责每次从数据流里取下一个值。今天这篇是 Python 3.12 魔术方法系列的第 95 讲我围绕__anext__展开聊透它到底是什么、怎么做标准实现、适合哪些场景以及我这些年踩过的几个坑。这个方法的定位其实很朴素它是异步世界里的__next__。同步迭代器靠__next__拿下个元素异步迭代器靠__anext__拿下个元素差别在于__anext__返回的是可等待对象而不是现成的值。为什么非要这样设计因为你在异步迭代中等的东西通常是网络字节、队列消息、文件流、数据库游标这类慢速资源。如果__anext__是同步阻塞的async for一遇到慢 I/O 就把整个事件循环卡死异步编程最重要的并发调度就全废了。它必须能被await事件循环才能在等待期间切去跑别的任务。1. 先搞清异步迭代器协议在解决什么问题1.1 从__next__到__anext__同步与异步的完整对照要理解__anext__最好的办法是把它放进一个对照表里看。同步迭代器和异步迭代器在协议层面长得几乎一模一样只是一个没有await一个有await维度同步迭代器异步迭代器获取迭代器__iter____aiter__取下一个元素__next____anext__结束信号StopIterationStopAsyncIteration单次返回值直接返回元素返回 awaitableawait后得到元素消费语法forasync for这表格里每一个格子都是成对出现的。你不能只写__aiter__却没有__anext__也不能只写__anext__却没有__aiter__async for在协议层会同时检查两者。实际开发里最容易漏的恰恰是只写了返回self的__aiter__然后幻想着async for会自动把一个普通方法变成协程。它不会。同步世界里还有个细节常被忽略for循环能直接消费实现了__iter__的对象也能消费生成器因为生成器内部同时实现了__iter__和__next__。异步世界也一样异步生成器同时实现了__aiter__和__anext__。这也是为什么你写个async def生成器函数后直接async for就能用完全不用手动补协议方法——解释器在背后帮你把这两个方法都生成好了。1.2 协议三件套__aiter__、__anext__、StopAsyncIteration__aiter__是入口它是一个同步方法返回一个异步迭代器对象__anext__是取数方法通常写成async defStopAsyncIteration是终止信号用于告诉消费者“没有下一个了”。这三者共同构成一个完整的异步迭代器协议缺一不可。终止信号用异常而不是返回值来表达这是 Python 一个很深思熟虑的设计。它让“本轮没取到数据”和“数据流已经结束”这两个概念彻底分开。举个生活化的例子排队取号时窗口叫“下一个”是正常业务喇叭广播“今日号已取完”是业务流程结束。如果都用返回值表达取到None到底是没号了还是服务员暂时卡了一下说不清。异常专门承担“停止”这个语义干净利落也不会污染正常的业务返回值。另外要注意StopAsyncIteration和StopIteration在异步协议里不能混用。async for的解释器实现捕获的是StopAsyncIteration你如果抛了StopIteration循环不会把它当作结束信号处理而是当作一个普通异常冒泡出去直接让你的任务崩溃。这一点后面专门展开但先有个印象异步代码里停止迭代必须用StopAsyncIteration。1.3async for到底被解释器改写成什么很多教材讲async for只讲语法不讲展开形式。但把展开形式看清楚你才能真正理解__anext__的角色以及每种报错是从哪来的。# 你写的代码 async for item in stream: handle(item) # Python 解释器大体会按这个逻辑处理 _it stream.__aiter__() while True: try: item await _it.__anext__() except StopAsyncIteration: break else: handle(item)注意两个细节。第一__aiter__()是同步调用前面没有await它必须直接返回一个异步迭代器对象。第二__anext__()的返回值必须被await所以这个方法返回的东西必须是一个 awaitable。常见的坑就在这里如果你在__anext__里直接return 1那解释器执行到await _it.__anext__()时相当于await 1会直接抛出TypeError: object int cant be used in await expression。搞清楚这个展开逻辑之后很多报错就变得可推理了。你说一个对象“不能用于 async for”不外乎三种原因没有__aiter__、__aiter__没有返回迭代器、或者返回的迭代器没有__anext__。排查的时候按这个顺序来基本不会白费功夫。2. 手写一个真正的异步迭代器2.1 最小可运行实现从 1 数到 3先给一个最直接、最能跑通的示例。一个异步计数器每次取一个数到上限就抛出StopAsyncIterationimport asyncio from typing import AsyncIterator class AsyncCounter: def __init__(self, limit: int): self._current 0 self._limit limit def __aiter__(self) - AsyncCounter: return self async def __anext__(self) - int: if self._current self._limit: raise StopAsyncIteration # 模拟一个真实的异步等待点 await asyncio.sleep(0.01) self._current 1 return self._current async def main(): async for value in AsyncCounter(3): print(value) asyncio.run(main())逐行拆解一下。__aiter__是普通方法返回self这是个很常见的写法迭代器本身就是可迭代对象一次迭代用完。__anext__用async def声明这是一个协程函数调用它得到的是协程对象天然满足“返回 awaitable”的要求。内部先检查状态到上限就抛终止异常否则模拟一个异步等待然后状态加一、返回值。这个例子里await asyncio.sleep(0.01)看起来很刻意但它承担了一个关键职责让__anext__有一个真正的挂起点。在实际项目里这个位置往往是await redis.get()、await queue.get()、await socket.read()之类的真实 I/O。你可以把asyncio.sleep替换成任何异步调用模式不变。2.2 状态管理与终止信号每个分支都要收敛迭代器是有状态的。同步迭代器靠实例属性记录“下一步是几”异步迭代器也一样通常用一个或多个实例字段保存当前游标、结束标记、内部缓冲区。写__anext__时一个非常重要的审查标准是方法里的每个分支都必须收敛——要么返回值要么抛StopAsyncIteration要么抛业务异常。最怕的是“隐式return None”。比如你写了一个if ...: return ...但条件不满足时没有 else函数就直接走到末尾返回None。async for那边还在傻傻await None立刻弹出TypeError。这种报错不像缺少协议方法那样直白排查时容易绕圈。我自己的习惯是把状态变化尽量集中。以一个分页拉取为例页数、是否还有下一页、当前是否已关闭这几个状态最好是显式字段并且每次__anext__只在一个地方推进游标。如果你在多个分支里跳跃地改状态很容易出现“明明还有数据却不继续迭代”或“循环根本停不下来”的边界问题。2.3__aiter__为什么不能写成异步函数这个问题几乎每个写自定义异步迭代器的人都问过__aiter__能不能加个async变成一个协程答案是明确的不能。协议要求__aiter__是一个普通同步方法调用后立即得到一个异步迭代器。如果你写成async def __aiter__那async for执行stream.__aiter__()时拿到的是一个协程对象而不是迭代器对象。解释器期望的是“可以继续调__anext__”的东西现在等在那里的却是一个还没跑的协程整个协议就断裂了。实际报错通常是TypeError: object cant be used in async for expression。这个错很有迷惑性因为它不会直接告诉你“__aiter__不能是异步函数”。所以我把这个坑放在前面写任何异步迭代器类先检查__aiter__是不是普通def不是再去查别的。2.4 类型注解与标准抽象基类的配合在 Python 3.12 里类型标注已经非常成熟异步迭代器也有对应的标准抽象基类collections.abc.AsyncIterable和collections.abc.AsyncIterator。前者代表“可以被异步迭代”后者代表“是一个异步迭代器”。from collections.abc import AsyncIterable, AsyncIterator def consume_batch(items: AsyncIterable[str]) - None: async for item in items: process(item) def fetch_page_stream() - AsyncIterator[str]: return some_async_generator()这里有个容易混淆的点AsyncIterable不一定能多次迭代。如果你的__aiter__每次返回一个新的迭代器对象那这个对象就可以被反复async for如果__aiter__只是返回self那它本质上是一次性可迭代对象。类型注解表达不了这种语义区别所以你需要在注释或文档里写清楚避免调用方误用。我以前遇到过一个问题把__aiter__返回类型标成AsyncIterator[int]但方法体里返回了self而self确实实现了协议类型检查器没有报错逻辑也正常。可一旦后续重构把__anext__删了类型检查器立刻就会报警因为它发现类的接口不满足AsyncIterator。这就是用抽象基类做标注的价值把协议约束提前到静态检查阶段。3. 换到真实场景里队列、分页与生成器偷懒3.1 场景一监听异步队列直到哨兵出现异步队列是__anext__最典型的应用场景之一。消息队列的消费端本质上就是“不断取下一个消息取不到就挂起等待”。很多框架的队列对象已经实现了异步迭代协议你可以直接写async for item in queue。但如果你需要定制结束条件最通用的做法是自己包一层。下面这个QueueReader演示了哨兵模式队列里放入一个特殊对象消费端遇到它就知道该结束了。import asyncio class QueueReader: def __init__(self, queue: asyncio.Queue, sentinel: object): self._queue queue self._sentinel sentinel def __aiter__(self): return self async def __anext__(self): item await self._queue.get() if item is self._sentinel: raise StopAsyncIteration return item使用的时候生产者往队列里塞数据最后塞一个哨兵async def producer(queue: asyncio.Queue): for i in range(5): await queue.put(i) await queue.put(None) # sentinel, 实际项目中可以用一个专属对象 async def consumer(queue: asyncio.Queue): async for item in QueueReader(queue, None): print(item)这个模式在实际项目里很常见。哨兵对象尽量避免用None因为你的业务数据里可能真的会出现None。更好的做法是定义一个独一无二的哨兵实例_SENTINEL object()。这样即使队列里出现过None也不会误伤。还有一个容易忽略的点队列本身不会自动结束。如果你不用哨兵async for会一直阻塞在读队列上因为__anext__内部的await queue.get()是永不结束的等待。这就是异步迭代器与普通迭代器的差异普通列表迭代完了自然抛StopIteration但如果__anext__里是无限等待的调用那这个迭代器从语义上就是无穷流。无穷流本身没问题问题是你必须设计一个外部手段去终止它哨兵就是最朴素可靠的手段。3.2 场景二远端分页拉取一个经典的有限异步迭代器另一个常见需求是把“分页拉取远端数据”封装成异步迭代器。比如对接某个接口每次拿一页接口告诉你有没有下一页然后你像遍历本地列表一样遍历所有页。class PaginatedFetcher: def __init__(self, client, page_size: int 20): self._client client self._page_size page_size self._page 0 self._closed False def __aiter__(self): return self async def __anext__(self): if self._closed: raise StopAsyncIteration data, has_next await self._client.fetch(self._page, self._page_size) self._page 1 if not has_next: self._closed True return data这个实现的核心技巧是用_closed标志统一管理结束状态。第一次发现没有下一页时把_closed置成True下一次调用直接抛StopAsyncIteration。这样写比在__anext__里记一个“是否还有下一页”的状态更稳因为无论调用方是提前 break 还是完整遍历完状态都是一致的。分页场景里还有一个业务细节要注意如果你在await self._client.fetch(...)这一步抛出了网络异常异常会直接传给消费者此时_page已经自增了下次重试就会跳到下一页。这在某些业务里是对的但在另一些业务里你可能希望暂停重试而不是跳页。具体怎么处理取决于你对数据完整性要求有多高我在实际项目里更倾向于在迭代器外部做重试让迭代器本身保持“一次取一页失败就抛错”的纯净语义。3.3 场景三用async generator偷懒让协议留在幕后手写协议方法虽然清晰但要说最舒服的写法还得是异步生成器。它让你只关注“怎么产生数据”完全不用管__aiter__和__anext__的细节。async def fetch_pages(client, page_size: int 20): page 0 while True: data, has_next await client.fetch(page, page_size) yield data if not has_next: return page 1这个生成器函数被调用后得到的是一个异步生成器对象它自动实现了__aiter__和__anext__。你在async for里消费它时解释器每次都会调用底层的__anext__驱动函数体执行到下一个yield。不要因为async generator用得爽就完全忘了__anext__的存在。理解底层机制依然重要生成器内部的return会被解释器转换成StopAsyncIteration抛给消费者yield则相当于__anext__返回了值。这种对应关系能帮你理解很多奇怪的行为比如为什么在生成器里写raise StopIteration会被转成RuntimeError而return才是正规结束方式。3.4 手写类和异步生成器的取舍当你有多个可选方案时选择标准很简单需要精细控制资源生命周期就用手写类只是单纯消费一个异步数据流做变换就用异步生成器。需求特征推荐方案快速原型、临时脚本异步生成器需要自定义aclose()做清理手写类迭代器要维护多个状态字段手写类对已有异步函数做流式包装异步生成器要参与类型抽象、被别的框架识别手写类并实现AsyncIterator我个人的默认方案是第一版先用异步生成器跑通业务逻辑确认数据流形态正确之后如果发现确实需要额外的状态管理或资源清理再改造成手写类。过早设计成手写类往往会带来不必要的复杂状态。4.anext()内置函数与高频坑位4.1anext()的三种常见用法Python 3.10 引入了内置函数anext()作用同next()完全对应。既然你在用 Python 3.12这个函数就要高频用起来。value await anext(iterator) # 拿下一个拿不到抛 StopAsyncIteration value await anext(iterator, default_value) # 拿不到时返回默认值 try: value await anext(iterator) except StopAsyncIteration: # 手动处理结束 ...带默认值的形式特别适合判断流是否结束。比如你只想检查一下异步迭代器是否还有数据又不想写一长串try/exceptfirst_item await anext(stream, None) if first_item is None: print(stream is empty) else: print(ffirst item: {first_item})这种写法在消费“头部数据”或“只处理第一条消息”的场景里很好用。但要注意anext(stream, None)只有在数据流真的结束时才会返回默认值如果数据流里的第一个元素恰好是None那和“没有元素”无法区分。所以选择默认值时一定要想清楚业务语义。4.2 坑一__anext__必须返回 awaitable而不是值本身最容易踩的坑就是用同步函数的思维写__anext__class WrongIterator: def __init__(self, data): self._data data self._i 0 def __aiter__(self): return self def __anext__(self): # 这是错的 if self._i len(self._data): raise StopAsyncIteration value self._data[self._i] self._i 1 return value # 返回了普通 int表面的报错可能很突然TypeError: object NoneType cant be used in await expression或者更直接地提示你不能 await 一个整数。原因就是协议要求__anext__返回一个 awaitable而return value返回的是实际元素。解决办法有两个要么把__anext__定义成async def让返回值自动变成协程然后在函数内部return value要么保持普通def但手动返回一个 awaitable比如return asyncio.sleep(0, resultvalue)。后者非常绕基本没必要。我建议一律用async def语义清晰还能防止你手滑返回普通值。4.3 坑二StopAsyncIteration和StopIteration不能混用这个坑我刚才提过这里展开说透。在async for的展开代码里只有StopAsyncIteration会被当作正常结束信号。如果你抛的是StopIteration对解释器来说这就是一个普通异常会直接穿过循环冒泡到外层。比如这个写法就是典型的错误async def __anext__(self): if self._closed: raise StopIteration # 错应该是 StopAsyncIteration在异步生成器里问题更加隐蔽。如果你在async def生成器函数内部显式写raise StopIterationPython 会把 PEP 479 的规则同样应用过来直接抛一个RuntimeError: async generator raised StopIteration。也就是说不管是手写类还是异步生成器异步场景里都不要显式抛StopIteration。停止迭代的统一做法是手写类抛StopAsyncIteration异步生成器用return。4.4 坑三多个任务并发消费同一个异步迭代器__anext__是带状态的。两个并发的消费者任务同时调用同一个迭代器实例的__anext__本质上是同时读写同一个实例字段结果很容易是重复消费、漏消费、甚至游标错乱。举个例子如果你把PaginatedFetcher同时交给两个 worker 去消费async with asyncio.TaskGroup() as tg: tg.create_task(worker(fetcher)) tg.create_task(worker(fetcher))两个 worker 可能拿到同一页数据也可能互相覆盖_page状态行为完全随机。异步迭代器本身不是线程安全的也不应该默认是并发安全的。如果你真的需要并行消费通常做法是让每个 worker 持有独立的迭代器或者改用“批量取出再分发”的模式。比如一次从数据流里取出 N 个元素然后开任务并行处理这批元素。这比让多个 worker 抢同一个迭代器要可控得多。4.5 坑四任务取消与资源清理async for结束只会捕获StopAsyncIteration它不会自动帮你关闭底层资源。如果你在__anext__里打开了一个文件、一个连接、或者一个订阅通道而消费者提前break退出底层资源并不会被自动释放。一个稳妥的设计是给你的迭代器实现aclose()并在消费方用async with或try/finally保证资源的收尾class SubscriptionIterator: def __init__(self, conn): self._conn conn self._closed False def __aiter__(self): return self async def __anext__(self): if self._closed: raise StopAsyncIteration try: return await self._conn.receive() except EndOfStream: self._closed True raise StopAsyncIteration async def aclose(self): if not self._closed: self._closed True await self._conn.close()使用方把aclose()放进contextlib.aclosing()或者try/finally里这样无论正常结束还是提前 break连接都能被关闭。还有一个和任务取消相关的小知识点asyncio.CancelledError本质上也是一个异常如果任务在await self._conn.receive()时被取消这个异常会往上冒。除非你有明确的清理逻辑否则不建议在__anext__里吞掉它。你的迭代器越“透明”上层越容易正确接管异常处理。5. 怎么验证自己的__anext__写得对5.1 最小验证脚本用asyncio.run快速跑通不写完整测试前先用一个小脚本验证行为是最快的闭环方式。针对上文的AsyncCounter可以这样快速检查import asyncio async def main(): counter AsyncCounter(3) print(await anext(counter)) # 1 print(await anext(counter)) # 2 print(await anext(counter)) # 3 print(await anext(counter, END)) # END因为已经耗尽 asyncio.run(main())这个脚本能一次性验证几个关键点正常取数对不对、终止信号能不能被正确捕获、默认值在耗尽时是否生效。我每次写完手写__anext__都会先用这种 10 行以内的小脚本验证再补正式单测。原因很简单单测跑起来有框架开销调试信息可能被吞掉而小脚本直接打印一眼就能看清楚问题出在哪个环节。5.2 标准库单测模板三条核心路径必须覆盖验证__anext__时我至少覆盖三条路径正常迭代到耗尽、中途提前 break、异常传播。这三条路径正好对应了消费者最常见的三种使用方式。用标准库的unittest.IsolatedAsyncioTestCase就能组织得很清楚import unittest class AsyncCounterTest(unittest.IsolatedAsyncioTestCase): async def test_normal_iteration(self): result [value async for value in AsyncCounter(3)] self.assertEqual(result, [1, 2, 3]) async def test_exhaust_after_limit(self): counter AsyncCounter(1) self.assertEqual(await anext(counter), 1) with self.assertRaises(StopAsyncIteration): await anext(counter) async def test_exception_propagation(self): class Broken(AsyncCounter): async def __anext__(self): raise RuntimeError(boom) stream Broken(1) with self.assertRaises(RuntimeError): await anext(stream) if __name__ __main__: unittest.main()test_normal_iteration验证主流程test_exhaust_after_limit验证终止test_exception_propagation验证业务异常会正常向外抛而不是被async for吞掉。这三条跑通基本说明协议本身没有问题了。如果你想再严格一点加一条“提前 break 后再次迭代”的测试。注意这取决于你的实现语义如果__aiter__返回的是同一个self那 break 之后再async for迭代器可能已经处于耗尽状态。要么文档写清楚“一次性迭代”要么实现成每次都返回新的迭代器两类需求对测试要求完全不同。6. 我写异步迭代器这几年沉淀下来的习惯6.1 现在的默认方案能生成器不手写必须手写时状态做减法我现在的默认方案已经非常固定如果只是给已有异步数据流做转换、分页、过滤优先用异步生成器让解释器负责协议细节如果这个对象要承担资源管理、状态机职责或者要被框架注册成某种异步数据源才手写类。手写类时我会刻意做状态减法。实例字段只保留“是否已关闭”和“当前游标”两类其余信息尽量从下层对象拿。很多复杂的 bug根源都是状态字段太多导致__anext__的不同分支对状态的理解不一致。把状态集中在一两个标志上分支逻辑会简单很多。6.2 最后一个调试小技巧打开 asyncio 的 debug 模式如果你发现async for卡住不动或者某个等待点迟迟不恢复先别急着加日志。用asyncio.run(main(), debugTrue)跑一遍事件循环会把每个协程的创建位置、当前等待对象、阻塞时间输出出来。__anext__内部的协程栈往往一清二楚是卡在queue.get()上还是卡在一次网络调用上很快就能定位。如果你发现迭代“提前退出”优先检查__anext__是否在某个分支无条件抛了StopAsyncIteration如果你发现迭代根本不结束优先检查__anext__内是否有分支漏掉了终止信号的抛出。这两个问题方向正好相反排查逻辑也正好对称。把这段经验留下来希望能帮你少走几圈弯路。
