深入异步迭代器：从 aiter/anext 到手写 Async Iterator 的实战指南
在异步编程日益普及的今天，理解 Python 的异步迭代器（
__aiter__/__anext__）机制，不仅能提升你对协程的掌控力，更是构建高性能异步系统的关键一环。本文将带你从基础语法出发，逐步手写一个完整的异步迭代器，深入理解其运行原理与实战应用。
一、为什么要关注异步迭代器？
在传统同步代码中，我们习惯使用for循环遍历可迭代对象（如列表、生成器等）。但在异步编程中，数据往往是“延迟到达”的，比如：
- 网络请求返回的数据流（如 WebSocket、API 分页）
- 异步文件读取（如 aiofiles）
- 实时传感器数据流
这类场景下，使用同步迭代器会阻塞事件循环，影响性能。而异步迭代器（Async Iterator）正是为此而生，配合async for实现非阻塞的数据消费。
二、异步迭代器的基本语法与原理
1. 关键方法：__aiter__与__anext__
要实现一个异步迭代器，需要定义两个魔法方法：
classAsyncCounter:def__init__(self,limit):self.limit=limit self.current=0def__aiter__(self):returnselfasyncdef__anext__(self):ifself.current<self.limit:awaitasyncio.sleep(1)# 模拟异步操作self.current+=1returnself.currentelse:raiseStopAsyncIteration使用方式：
importasyncioasyncdefmain():asyncfornuminAsyncCounter(3):print(num)asyncio.run(main())输出：
1 2 32. Python 内置函数：aiter()与anext()
从 Python 3.10 起，官方引入了两个新函数：
aiter(obj)：等价于obj.__aiter__()，返回异步迭代器。anext(iterator, default)：等价于await iterator.__anext__()，支持设置默认值。
示例：
it=aiter(AsyncCounter(2))print(awaitanext(it))# 输出 1print(awaitanext(it))# 输出 2print(awaitanext(it,'Done'))# 输出 'Done'，不抛异常三、手写一个异步数据源模拟器
我们以“异步日志流读取器”为例，模拟一个每秒产生一条日志的异步数据源，并实现异步迭代器接口。
1. 需求分析
- 每秒生成一条日志（模拟异步数据流）
- 支持
async for遍历 - 支持外部取消（如读取到某条日志后终止）
2. 实现代码
importasyncioimportrandomfromtypingimportOptionalclassAsyncLogStream:def__init__(self,max_lines:int=10):self.max_lines=max_lines self.count=0def__aiter__(self):returnselfasyncdef__anext__(self):ifself.count>=self.max_lines:raiseStopAsyncIterationawaitasyncio.sleep(random.uniform(0.5,1.5))# 模拟不稳定的网络延迟self.count+=1returnf"[LOG-{self.count}] System heartbeat OK"3. 消费者代码
asyncdefmonitor_logs():asyncforlineinAsyncLogStream(5):print(line)if"3"inline:print("⚠️ 检测到关键日志，提前终止")breakasyncio.run(monitor_logs())输出示例：
[LOG-1] System heartbeat OK [LOG-2] System heartbeat OK [LOG-3] System heartbeat OK ⚠️ 检测到关键日志，提前终止四、异步迭代器 VS 同步生成器：性能与适用场景对比
| 特性 | 同步生成器 | 异步迭代器 |
|---|---|---|
| 关键语法 | __iter__/__next__ | __aiter__/__anext__ |
| 使用方式 | for item in gen: | async for item in agen: |
| 是否阻塞 | 是 | 否（非阻塞） |
| 适用场景 | 本地数据、CPU 密集型 | 网络 I/O、实时数据流 |
| 示例库 | itertools | asyncio、aiohttp、aiokafka |
五、进阶技巧：异步生成器（async def + yield）
除了手写__aiter__/__anext__，Python 还支持更简洁的异步生成器语法：
asyncdefasync_counter(limit):foriinrange(limit):awaitasyncio.sleep(1)yieldiasyncdefmain():asyncforiinasync_counter(3):print(i)asyncio.run(main())这背后其实是 Python 自动帮你实现了异步迭代器协议。
六、实战案例：异步分页 API 抓取器
场景：
你需要从一个分页 API 异步抓取数据，每页最多返回 100 条记录，直到返回空列表为止。
实现：
classAsyncPaginator:def__init__(self,fetch_page_func):self.fetch_page=fetch_page_func self.page=1def__aiter__(self):returnselfasyncdef__anext__(self):data=awaitself.fetch_page(self.page)ifnotdata:raiseStopAsyncIteration self.page+=1returndata模拟 API：
asyncdefmock_fetch_page(page):awaitasyncio.sleep(0.5)ifpage>3:return[]return[f"item-{page}-{i}"foriinrange(5)]使用方式：
asyncdefmain():asyncforpage_datainAsyncPaginator(mock_fetch_page):print("📦 收到数据：",page_data)asyncio.run(main())七、最佳实践与建议
✅ 推荐：
- 使用
async for替代手动anext()，更安全简洁。 - 异步迭代器适合处理“流式数据”或“分页数据”。
- 尽量使用
async def + yield简化代码。 - 使用
aiter()/anext()提升代码兼容性与可读性。
⚠️ 注意：
- 异步迭代器不能用于普通
for循环。 __anext__必须是async def，否则会抛出TypeError。- 异步迭代器不支持
break后自动关闭（如需清理资源，需配合aclose()）。
八、未来展望：异步数据流的主战场
随着微服务、实时数据处理、AI 推理流等场景的兴起，异步迭代器将成为 Python 异步生态的重要基石。无论是构建高性能爬虫、流式数据处理框架，还是异步数据库驱动（如 asyncpg、motor），都离不开对异步迭代协议的深入理解。
九、总结与互动
本文从底层协议出发，手写了多个异步迭代器示例，帮助你理解__aiter__/__anext__的运行机制，并结合实际场景展示了其强大威力。
你是否在项目中使用过异步迭代器？你更喜欢手写协议，还是使用 async generator？欢迎在评论区分享你的经验与思考！
附录与参考资料
- PEP 492 – Coroutines with async and await syntax
- PEP 525 – Asynchronous Generators
- Python 官方文档：Asynchronous Iterators (docs.python.org in Bing)
- 推荐书籍：《流畅的 Python》《异步 Python 编程实战》
