如何将AsyncIterator转换为普通同步Iterator？

结论 #

可以通过封装的方式得到懒加载的同步迭代器，无需提前将所有异步迭代元素转换为列表存储。

现有实现的问题 #

你当前的sync_all方法属于一次性全量消费实现：

• 优点：逻辑简单，适合小体量异步迭代场景
• 缺点：必须等待所有异步元素拉取完成才能开始迭代，遇到大数据集、无限长度的异步迭代器会占用过高内存，甚至触发OOM

懒加载封装实现 #

核心逻辑：同步迭代器每次调用__next__方法时，仅在事件循环中执行一次异步迭代器的__anext__操作，拿到当前元素后直接返回，无需存储全量结果。

完整实现代码 #

import asyncio
from typing import Generic, TypeVar, AsyncIterable, Iterable

T = TypeVar("T")

class AsyncToSyncIterator(Generic[T]):
    def __init__(self, async_iterable: AsyncIterable[T]):
        self.async_iter = async_iterable.__aiter__()
        # 事件循环初始化：优先复用当前线程现有循环，不存在则新建
        try:
            self.loop = asyncio.get_event_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self.loop)

    def __iter__(self) -> Iterable[T]:
        return self

    def __next__(self) -> T:
        try:
            # 单次迭代仅执行一次异步拉取
            return self.loop.run_until_complete(self.async_iter.__anext__())
        except StopAsyncIteration:
            # 异步迭代结束，转换为同步迭代终止异常
            self.loop.run_until_complete(self.loop.shutdown_asyncgens())
            self.loop.close()
            raise StopIteration

# 工具类封装
class AsyncToSyncIterableUtils(Generic[T]):
    @staticmethod
    def sync_iter(iterable: AsyncIterable[T]) -> Iterable[T]:
        """返回懒加载的同步迭代器"""
        return AsyncToSyncIterator(iterable)

    @staticmethod
    def sync_all(iterable: AsyncIterable[T]) -> list[T]:
        """返回全量结果列表，等价于你原有实现"""
        return list(AsyncToSyncIterator(iterable))

使用示例 #

# 测试用异步生成器
async def async_demo_gen(count: int):
    for i in range(count):
        await asyncio.sleep(0.1)
        yield f"异步生成元素：{i}"

# 懒加载迭代：不需要等待所有元素生成完成，拉取到一个就处理一个
for item in AsyncToSyncIterableUtils.sync_iter(async_demo_gen(5)):
    print(item)

注意事项 #

• 该实现仅支持同步上下文环境使用，如果你当前代码本身就在运行事件循环（如Jupyter环境、异步Web服务请求上下文），会触发事件循环重复运行的冲突，该场景下建议直接使用异步迭代逻辑
• 无限长度的异步迭代器可以正常使用，内存占用恒定，不会出现内存溢出问题

内容的提问来源于stack exchange，提问作者Dunes Buggy