你在写异步代码的时候，碰到过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：

这表格里每一个格子都是成对出现的。你不能只写__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 def；StopAsyncIteration是终止信号，用于告诉消费者“没有下一个了”。这三者共同构成一个完整的异步迭代器协议，缺一不可。

终止信号用异常而不是返回值来表达，这是 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 can't be used in 'await' expression。

搞清楚这个展开逻辑之后，很多报错就变得可推理了。你说一个对象“不能用于 async for”，不外乎三种原因：没有__aiter__、__aiter__没有返回迭代器、或者返回的迭代器没有__anext__。排查的时候按这个顺序来，基本不会白费功夫。

2. 手写一个真正的异步迭代器

2.1 最小可运行实现：从 1 数到 3

先给一个最直接、最能跑通的示例。一个异步计数器，每次取一个数，到上限就抛出StopAsyncIteration：

import 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 can't 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 手写类和异步生成器的取舍

当你有多个可选方案时，选择标准很简单：需要精细控制资源生命周期，就用手写类；只是单纯消费一个异步数据流做变换，就用异步生成器。

我个人的默认方案是：第一版先用异步生成器跑通业务逻辑，确认数据流形态正确之后，如果发现确实需要额外的状态管理或资源清理，再改造成手写类。过早设计成手写类，往往会带来不必要的复杂状态。

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/except：

first_item = await anext(stream, None) if first_item is None: print("stream is empty") else: print(f"first 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 can't be used in 'await' expression，或者更直接地提示你不能 await 一个整数。原因就是协议要求__anext__返回一个 awaitable，而return value返回的是实际元素。

解决办法有两个：要么把__anext__定义成async def，让返回值自动变成协程，然后在函数内部return value；要么保持普通def，但手动返回一个 awaitable，比如return asyncio.sleep(0, result=value)。后者非常绕，基本没必要。我建议一律用async def，语义清晰，还能防止你手滑返回普通值。

4.3 坑二：StopAsyncIteration和StopIteration不能混用

这个坑我刚才提过，这里展开说透。在async for的展开代码里，只有StopAsyncIteration会被当作正常结束信号。如果你抛的是StopIteration，对解释器来说这就是一个普通异常，会直接穿过循环冒泡到外层。

比如这个写法就是典型的错误：

async def __anext__(self): if self._closed: raise StopIteration # 错！应该是 StopAsyncIteration

在异步生成器里，问题更加隐蔽。如果你在async def生成器函数内部显式写raise StopIteration，Python 会把 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(), debug=True)跑一遍，事件循环会把每个协程的创建位置、当前等待对象、阻塞时间输出出来。__anext__内部的协程栈往往一清二楚：是卡在queue.get()上，还是卡在一次网络调用上，很快就能定位。

如果你发现迭代“提前退出”，优先检查__anext__是否在某个分支无条件抛了StopAsyncIteration；如果你发现迭代根本不结束，优先检查__anext__内是否有分支漏掉了终止信号的抛出。这两个问题方向正好相反，排查逻辑也正好对称。把这段经验留下来，希望能帮你少走几圈弯路。