如何在Python中为非异步函数编写异步包装器？

当你遇到外部库提供的无await的同步长耗时函数时，核心问题是这类函数会阻塞asyncio的单线程事件循环——一旦调用它，整个事件循环会被卡住，无法处理其他协程。解决思路是把同步函数放到线程池或进程池里执行，让事件循环在等待结果时可以调度其他任务。

下面是两种实用的实现方式：

方法1：使用asyncio.to_thread()（Python 3.9+ 推荐） #

这是Python 3.9新增的官方方法，专门用于快速将同步函数包装为异步接口，底层自动使用线程池。

示例代码 #

首先定义你的同步长耗时函数：

def long_function():
    # 模拟实际场景中的长耗时操作（比如文件IO、网络请求、复杂计算）
    import time
    time.sleep(5)
    return "耗时任务执行完成"

然后编写异步包装器：

import asyncio

async def async_long_function():
    # 直接用to_thread包装同步函数，await等待线程执行结果
    return await asyncio.to_thread(long_function)

验证异步行为的主函数：

async def main():
    # 同时启动两个异步任务，验证事件循环不会被阻塞
    task1 = asyncio.create_task(async_long_function())
    task2 = asyncio.create_task(async_long_function())

    print("等待任务完成...")
    result1 = await task1
    result2 = await task2
    print(result1, result2)

asyncio.run(main())

运行后会看到两个任务并行执行，总耗时约5秒（而非10秒），和原生异步函数的行为一致。

方法2：使用loop.run_in_executor()（兼容低版本Python） #

如果你的Python版本低于3.9，可以用事件循环的run_in_executor()方法，手动指定执行器（线程池或进程池）。

线程池示例（IO密集型任务首选） #

import asyncio
from concurrent.futures import ThreadPoolExecutor

def long_function():
    import time
    time.sleep(5)
    return "耗时任务执行完成"

async def async_long_function():
    # 获取当前运行的事件循环
    loop = asyncio.get_running_loop()
    # 用ThreadPoolExecutor执行同步函数，也可以自定义线程池大小（比如max_workers=4）
    with ThreadPoolExecutor() as executor:
        return await loop.run_in_executor(executor, long_function)

进程池示例（CPU密集型任务首选） #

如果你的同步函数是CPU密集型（比如大计算量），受GIL限制，线程池无法利用多核，这时可以用ProcessPoolExecutor：

import asyncio
from concurrent.futures import ProcessPoolExecutor

def long_function():
    # 模拟CPU密集型计算
    sum(range(10**8))
    return "CPU密集任务完成"

async def async_long_function():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as executor:
        return await loop.run_in_executor(executor, long_function)

关键注意事项 #

• 传递参数：如果同步函数需要参数，直接在包装时传入，比如await asyncio.to_thread(long_function, arg1, arg2, kwarg1=value1)。
• 避免跨上下文调用：同步函数运行在单独的线程/进程中，不要在里面调用asyncio的API（比如asyncio.create_task），否则会引发错误。
• 执行器复用：如果频繁调用包装后的异步函数，建议提前创建全局执行器（而非每次创建新的），减少线程/进程创建的开销。

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