Python利用Motor实现MongoDB流式数据读取 async for遍历游标

在Motor 3.0+版本中，`find()` 返回的 `AsyncIOMotorCursor` 完全支持通过 `async for` 直接进行异步遍历，但许多开发者容易陷入一个常见误区——误以为“获取游标必须先使用 `await`”，于是写出 `cursor = await collection.find(...)`。这条语句会立即执行查询，导致游标进入“已消耗”状态，后续再使用 `async for doc in cursor` 进行迭代时，就会抛出 `RuntimeError: async generator already exhausted` 错误。要避免此问题，只需记住：直接使用 `cursor = collection.find(...)`（不加 `await`），然后安心地用 `async for doc in cursor` 进行迭代。如果需要限制返回文档数量，应使用 `.limit(n)` 方法，而不是 `to_list(n)`——后者会将所有数据一次性加载到内存中，从而失去流式处理的优势。此外，游标的生命周期与当前 asyncio 任务绑定，若任务被取消，Motor 会自动清理未完成的网络请求，无需担心资源泄漏问题。

Motor游标默认不支持async for直接遍历？

这个标题本身是一个反问，答案其实很明确：支持，但需满足前提条件。除了前面提到的不能提前使用 `await` 之外，还需要注意游标超时问题。MongoDB 默认对空闲游标设置 10 分钟超时（`maxTimeMS` 并不影响此设置），一旦超时，服务端会销毁游标，后续的 `async for` 将抛出 `ExecutionTimeout` 或 `InvalidOperation: cursor does not exist` 异常。这并非 Motor 的缺陷，而是 MongoDB 的保护机制。解决方案如下： - 对于长时间运行的流式处理，可以显式禁用超时：`collection.find(..., no_cursor_timeout=True)`。 - 更稳妥的做法是配合 `max_await_time_ms`（仅适用于 tailable 游标）或在业务层实现心跳机制：每处理 N 条数据后调用 `await asyncio.sleep(0)` 让出控制权，避免事件循环长时间阻塞导致游标空闲超时。 - 务必不要在 `async for` 循环内执行同步阻塞操作（如 `time.sleep`、文件读写），这会导致游标空闲时间延长，增加超时风险。

如何避免游标超时或连接中断导致遍历失败？

本质上，需要理解游标是一种有状态的服务端资源，而非纯粹的 Python 迭代器。除了超时问题，还要注意在遍历过程中能否修改查询条件。答案是否定的。MongoDB 游标是前向只读的，Motor 的 `AsyncIOMotorCursor` 不支持 `skip()`、`rewind()` 或动态修改 `filter`。所谓“跳过前 N 条”，必须在执行 `find()` 之前通过 `.skip(N)` 指定；而“按条件中断”只能通过 Python 层面的 `break` 实现，但这不会通知 MongoDB 提前终止游标——服务端仍会继续拉取数据，直到游标耗尽或超时。 如果需要分页式流式消费，一个高效的替代方案是使用 `_id` 范围查询代替 `skip()`：记录上一批最后一条文档的 `_id`，下一次查询条件设为 `{'_id': {'$gt': last_id}}`，配合排序即可。若想实现“消费到某时间戳就停止”，建议将时间字段加入到 `sort` 和 `filter` 中，而不是依赖循环内的判断。另外注意：`async for` 每次迭代实际上是一次 `getMore` 请求（默认 batch size 为 101），并非每条文档都发送一个网络包，因此“流式”的粒度是批次而非单条文档。

async for 遍历时能否中途修改查询条件或跳过数据？

有时候 `async for` 看起来“卡住”了，循环迟迟没有输出。最常见的原因是查询没有命中索引，MongoDB 执行了全表扫描（COLLSCAN），而 Motor 在等待第一批 batch 返回。此时 `async for` 会挂起，直到服务端返回首个结果块，期间没有任何日志或超时提示。要快速诊断，可以使用 `explain()` 查看执行计划——`await collection.find({...}).explain()`，重点关注 `executionStats.nReturned > 0` 以及 `stage` 是否为 `IXSCAN`。如果看到 `COLLSCAN`，说明索引缺失。另外，确保 `sort()` 字段也有对应索引，否则即使有 `limit()`，游标也无法高效推进。调试阶段可以在循环内加入 `print(f"got {doc.get('_id')}")` 来观察进度，但生产环境中高频 print 会成为性能瓶颈，建议改用日志采样方式。

为什么有时 async for 看起来“卡住”没输出？

最后总结一句：Motor 的流式读取本质上是让异步 I/O 与游标分批拉取自然对齐。关键不在于语法有多炫酷，而在于准确理解游标的生命周期、服务端约束以及事件循环的协作边界。真正容易被忽略的是，将“数据库游标”当作“Python 迭代器”来使用——它拥有服务端状态、超时逻辑和资源释放规则，并非纯粹的协程语法糖。只要把握住这些要点，使用 `async for` 处理海量数据就能既简洁又高效。