在Python Asyncio中运行长耗时同步任务的正确方式

你的核心问题是同步的重计算任务阻塞了asyncio事件循环，导致消息接收/发送等异步操作无法正常进行。解决思路是把同步任务放到独立的线程池或进程池里执行，让asyncio事件循环可以继续处理其他任务，不会被阻塞。

核心方法：用asyncio.run_in_executor异步执行同步任务 #

这个API可以把同步函数提交到线程池或进程池，事件循环会异步等待任务完成，期间不会阻塞。

1. 最简单的线程池实现（默认） #

直接用asyncio内置的线程池，代码修改如下：

import asyncio
from typing import num

def heavy(i: num) -> num:
    # 这里是你的耗时重计算逻辑
    result = ...
    return result

async def main():
    while True:
        req = await getInput()
        # 把heavy任务丢到默认线程池执行，await等待结果
        result = await asyncio.get_event_loop().run_in_executor(None, heavy, req.num)
        await sendResult(result)

if __name__ == '__main__':
    asyncio.run(main())

• 第一个参数传None时，会使用asyncio默认的线程池。
• 这样heavy在后台线程跑，事件循环可以继续处理getInput()和sendResult()，不会阻塞消息通信。

2. 进程池实现（适合CPU密集型任务） #

如果你的heavy是纯CPU计算（比如大数据量运算），线程池因为GIL限制没法利用多核，建议用进程池：

import asyncio
from concurrent.futures import ProcessPoolExecutor
from typing import num

def heavy(i: num) -> num:
    # 耗时CPU密集型计算
    result = ...
    return result

async def main():
    # 创建进程池，max_workers建议设为CPU核心数
    executor = ProcessPoolExecutor(max_workers=4)
    try:
        while True:
            req = await getInput()
            # 用自定义进程池执行任务
            result = await asyncio.get_event_loop().run_in_executor(executor, heavy, req.num)
            await sendResult(result)
    finally:
        # 程序结束时关闭进程池
        executor.shutdown()

if __name__ == '__main__':
    asyncio.run(main())

• 进程池会创建独立的Python进程执行任务，每个进程有自己的GIL，能真正并行计算，大幅提升CPU密集型任务的效率。

关键注意点 #

• 线程池适合IO密集型的同步任务（比如调用同步IO接口），进程池适合CPU密集型任务。
• heavy函数必须是纯同步的，不能在里面调用asyncio的异步方法。
• 进程池的任务参数和返回值必须能被pickle序列化，否则会报错。

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