如何在Python asyncio中实现类似C# Task.Run的同步代码异步封装？

在Python的asyncio里，有两种常用方式可以实现类似C# Task.Run()的功能，把同步代码封装成可await的异步任务：

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

这个方法是专门为简化同步代码异步封装设计的，用法和Task.Run()一样直观，会自动把同步函数放到线程池执行，返回可await的对象：

import asyncio

# 定义同步耗时函数
def sync_heavy_task():
    print("开始执行同步任务")
    # 模拟IO/CPU操作
    import time
    time.sleep(2)
    return "同步任务执行完成"

async def main():
    # 包装同步函数并await结果
    task_result = await asyncio.to_thread(sync_heavy_task)
    print(task_result)

asyncio.run(main())

2. 使用loop.run_in_executor()（兼容Python 3.7+） #

如果你的Python版本低于3.9，可以用事件循环的run_in_executor()方法，它是to_thread()的底层实现，支持自定义线程/进程池：

import asyncio
from concurrent.futures import ThreadPoolExecutor

def sync_heavy_task():
    print("开始执行同步任务")
    import time
    time.sleep(2)
    return "同步任务执行完成"

async def main():
    loop = asyncio.get_running_loop()
    # 使用默认线程池执行同步任务
    result = await loop.run_in_executor(None, sync_heavy_task)
    
    # 也可以自定义线程池大小
    # with ThreadPoolExecutor(max_workers=4) as executor:
    #     result = await loop.run_in_executor(executor, sync_heavy_task)
    
    print(result)

asyncio.run(main())

额外说明 #

• 上述两种方式本质都是把同步代码放到线程池执行，避免阻塞asyncio的事件循环，和C# Task.Run()的线程池执行逻辑一致。
• 如果是CPU密集型的同步代码，可以用concurrent.futures.ProcessPoolExecutor代替线程池，绕过Python的GIL限制，但进程间通信开销更大，适合纯计算场景。

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