如何将AsyncIterable转换为同步Iterable传入同步函数？

我手头有两个对象：

• 一个接受同步Iterable作为输入的同步函数f
• 一个AsyncIterable类型的输入

需要在异步调用者中将后者转换成同步Iterable，才能传递给f使用。以下是示例代码及问题场景：

import asyncio


async def asquare(n):
    await asyncio.sleep(0.01 * n)
    return n * n


def first_even(numbers):
    # 该同步函数要求输入为同步Iterable
    return next(n for n in numbers if n % 2 == 0)


async def main(numbers):
    squares = ...  # 这里需要将异步生成的结果转为同步Iterable

    # 以下是几种尝试过但失败的写法：
    # 1. 异步生成器无法被同步函数迭代，抛出TypeError
    # squares = (await asquare(n) for n in numbers)

    # 2. 已存在运行中的事件循环时，asyncio.run()无法调用，抛出RuntimeError
    # squares = (asyncio.run(asquare(n)) for n in numbers)

    # 3. 不能在当前运行的事件循环中嵌套调用run_until_complete，抛出RuntimeError
    # squares = (asyncio.get_running_loop().run_until_complete(asquare(n)) for n in numbers)

    return first_even(squares)


if __name__ == "__main__":
    print(asyncio.run(main([1, 3, 5, 7, 9, 10, 11, 12, 13])))

注意要求 #

• 不能预迭代整个序列（比如用await asyncio.gather(...)），因为序列可能很长甚至是无限的
• 不使用nest_asyncio这类修改asyncio底层的方案
• 尽量避免额外线程或新事件循环，除非必须

解决方案 #

由于同步函数只能处理同步迭代逻辑，而异步结果必须在事件循环中获取，无法在当前运行的事件循环中嵌套执行异步任务，因此我们需要通过单独线程运行新事件循环的方式，实现异步结果的同步获取，且保证序列按需生成（不预迭代）。

实现转换工具函数

import asyncio
from threading import Thread
from typing import AsyncIterable, Iterable, TypeVar

T = TypeVar("T")

def async_to_sync_iter(async_iter: AsyncIterable[T]) -> Iterable[T]:
    """将AsyncIterable转换为按需生成的同步Iterable"""
    # 创建新的事件循环
    loop = asyncio.new_event_loop()

    def run_event_loop():
        # 在新线程中启动事件循环
        asyncio.set_event_loop(loop)
        loop.run_forever()

    # 启动守护线程运行事件循环
    thread = Thread(target=run_event_loop, daemon=True)
    thread.start()

    async def fetch_next():
        try:
            # 获取异步迭代器的下一个值
            return await anext(async_iter), False
        except StopAsyncIteration:
            # 迭代结束，停止事件循环
            loop.stop()
            return None, True

    while True:
        # 在当前线程中提交协程到新线程的事件循环，等待结果
        future = asyncio.run_coroutine_threadsafe(fetch_next(), loop)
        value, is_done = future.result()
        if is_done:
            break
        yield value

修改示例代码使用该工具

async def main(numbers):
    # 模拟AsyncIterable类型的输入（如果你的输入已经是AsyncIterable，可跳过此步）
    async def async_numbers():
        for n in numbers:
            yield n

    # 生成异步的平方序列
    async def async_squares():
        async for n in async_numbers():
            yield await asquare(n)

    # 转换为同步Iterable
    squares = async_to_sync_iter(async_squares())

    return first_even(squares)

方案说明 #

1. 按需生成：每次调用next()时才会获取异步迭代器的下一个值，不会预迭代整个序列，适配长序列或无限序列场景
2. 无补丁修改：没有修改asyncio的底层逻辑，完全基于官方API实现
3. 线程开销可控：仅使用一个守护线程运行事件循环，线程会在迭代结束后自动停止，不会造成资源泄漏

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