asyncio.wait_for无法按时终止含CPU密集操作的异步任务问题咨询

asyncio的wait_for超时机制只有在协程主动让出CPU控制权（比如遇到await关键字）时，事件循环才有机会检查是否超时。而time.sleep(3)是同步阻塞的CPU密集操作，会持续占用当前线程，事件循环被完全卡住，根本没法在超时时间到的时候终止协程，必须等这个同步操作执行完才能继续。

把CPU密集型的同步操作放到线程池中执行，通过await调用线程池任务，让协程主动让出CPU，这样事件循环就能正常检测超时。

方法1：使用asyncio.to_thread（Python 3.9+） #

这是Python3.9新增的便捷API，专门用来把同步函数放到线程池执行：

import asyncio
import time

def cpu_intensive_task():
    print("start CPU intensive task")
    time.sleep(3)
    print("end CPU intensive task")

async def f1():
    # 把CPU密集操作放到线程池执行，用await让出控制权
    await asyncio.to_thread(cpu_intensive_task)
    
    print("start asyncio.sleep")
    await asyncio.sleep(3)
    print("end asyncio.sleep")

async def main():
    print("-----f1-----")
    t1 = time.time()
    try:
        await asyncio.wait_for(f1(), timeout=2)
    except asyncio.TimeoutError:
        print("超时触发")
    t2 = time.time()
    print(f"f1 cost {(t2 - t1):.4f} s")

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

方法2：使用concurrent.futures.ThreadPoolExecutor（兼容低版本Python） #

如果用的是Python3.9以下版本，可以手动创建线程池：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

def cpu_intensive_task():
    print("start CPU intensive task")
    time.sleep(3)
    print("end CPU intensive task")

async def f1():
    loop = asyncio.get_running_loop()
    # 在线程池中执行同步任务
    await loop.run_in_executor(ThreadPoolExecutor(), cpu_intensive_task)
    
    print("start asyncio.sleep")
    await asyncio.sleep(3)
    print("end asyncio.sleep")

async def main():
    print("-----f1-----")
    t1 = time.time()
    try:
        await asyncio.wait_for(f1(), timeout=2)
    except asyncio.TimeoutError:
        print("超时触发")
    t2 = time.time()
    print(f"f1 cost {(t2 - t1):.4f} s")

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

修改后执行，会在2秒左右触发超时，输出类似：

-----f1-----
start CPU intensive task
超时触发
f1 cost 2.0012 s

这样就能实现整个f1在固定时间后终止，不管CPU密集操作有没有完成。

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