如何将Python生成器转换为异步生成器？IO密集场景适配方案

需求说明 #

现有一个IO密集型的Python同步生成器，希望将其转换为异步生成器，让生成器的循环逻辑运行在独立线程或进程中。例如从套接字加载数据块时，能在处理当前块的同时预加载下一块。计划通过队列让IO线程/进程缓冲生成器产出的结果，供异步生成器获取；优先使用concurrent.futures模块，以便灵活选择线程或进程实现。

示例同步代码 #

import time

def blocking():
    """ 带有阻塞IO的普通生成器 """
    i = 0
    while True:
        time.sleep(1)  # 模拟IO阻塞操作
        yield i  # 模拟生成的结果
        i += 1

def consumer():
    """ 同步消费者 """
    for chunk in blocking():
        print(chunk)

最优实现方案 #

核心思路是用concurrent.futures.Executor（线程/进程池）运行同步生成器，将产出结果存入asyncio.Queue，再通过异步生成器从队列中取数据，实现异步迭代。这种方案既保留了concurrent.futures的灵活性，又能通过队列实现缓冲，达到“处理当前块时预加载下一块”的效果。

代码实现 #

import asyncio
import concurrent.futures
import time

def blocking_generator():
    """ 带有阻塞IO的同步生成器 """
    i = 0
    while True:
        time.sleep(1)  # 模拟IO阻塞操作
        yield i
        i += 1
        # 测试用终止条件：生成5个结果后停止
        if i >= 5:
            break

async def async_generator_wrapper(executor, gen_func):
    """ 将同步生成器包装为异步生成器的工具函数 """
    # 设置队列缓冲大小，根据IO速度和处理速度调整
    queue = asyncio.Queue(maxsize=2)

    def producer():
        """ 在executor线程/进程中运行的生产者，负责读取生成器并填充队列 """
        try:
            for item in gen_func():
                # 同步线程中调用异步队列的put，需用run_coroutine_threadsafe
                asyncio.run_coroutine_threadsafe(queue.put(item), asyncio.get_event_loop()).result()
            # 生成器结束，放入None作为终止信号
            asyncio.run_coroutine_threadsafe(queue.put(None), asyncio.get_event_loop()).result()
        except Exception as e:
            # 将异常传入队列，由消费者处理
            asyncio.run_coroutine_threadsafe(queue.put(Exception(f"生产者出错: {str(e)}")), asyncio.get_event_loop()).result()

    # 提交生产者任务到executor
    future = executor.submit(producer)

    try:
        while True:
            item = await queue.get()
            if item is None:
                break  # 收到终止信号，停止迭代
            if isinstance(item, Exception):
                raise item  # 抛出生产者的异常
            yield item
            queue.task_done()
    finally:
        # 确保生产者任务被取消，避免资源泄漏（尤其针对无限循环生成器）
        future.cancel()

async def async_consumer():
    """ 异步消费者示例 """
    # IO密集型优先用ThreadPoolExecutor；若生成器含GIL阻塞的CPU操作，改用ProcessPoolExecutor
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        async for chunk in async_generator_wrapper(executor, blocking_generator):
            print(f"处理数据块: {chunk}")

if __name__ == "__main__":
    asyncio.run(async_consumer())

关键细节说明 #

1. 队列缓冲：asyncio.Queue(maxsize=2)控制缓冲数量，避免内存过载，同时保证生产者能提前预加载下一块数据。
2. 跨线程/进程通信：同步的生产者函数通过asyncio.run_coroutine_threadsafe将数据放入异步队列，解决了同步线程无法直接调用异步API的问题。
3. 终止与资源清理：生成器结束时放入None作为终止信号，异步生成器收到后停止迭代；finally块中取消executor任务，防止无限循环生成器导致的资源泄漏。
4. 灵活切换线程/进程：只需替换ThreadPoolExecutor为ProcessPoolExecutor即可切换为进程模式，注意进程模式下生成器的产出对象必须支持序列化（pickle）。

注意事项 #

• 若使用无限循环生成器，需在消费者侧添加终止逻辑（比如设置迭代次数、监听外部停止信号），否则生产者会持续运行。
• 进程池模式下，生成器函数和产出的所有对象都必须能被pickle序列化，否则会报错。
• 队列的maxsize需根据实际IO速度和处理速度调整：过小会导致生产者频繁等待，过大可能占用过多内存。

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