如何为含CPU与IO密集逻辑的async function设置全局超时？

你的核心问题是给同时包含CPU密集操作和异步IO操作的async function设置整体超时，之前的错误在于用run_in_executor直接调用异步函数——它只支持普通函数，导致协程未被执行的警告，同时CPU密集的time.sleep会阻塞事件循环，让异步逻辑无法正常工作。

正确的解决方案思路 #

1. 将CPU密集部分单独抽成普通函数，通过run_in_executor放到线程池执行，避免阻塞事件循环；
2. 原异步函数f1内部先await CPU密集任务的执行结果，再执行异步IO操作；
3. 直接用asyncio.wait_for包裹整个f1()调用，设置超时时间，实现对整个函数的超时控制。

修正后的代码 #

import asyncio
import time
import concurrent.futures

# 线程池用于执行CPU密集任务
executor = concurrent.futures.ThreadPoolExecutor(1)

# 抽离CPU密集操作到普通函数
def cpu_bound_task():
    print("start sleep")
    time.sleep(3)  # 模拟CPU密集型任务
    print("end sleep")

async def f1():
    loop = asyncio.get_running_loop()
    # 在线程池执行CPU密集任务，不阻塞事件循环
    await loop.run_in_executor(executor, cpu_bound_task)
    
    print("start asyncio.sleep")
    await asyncio.sleep(3)  # 模拟异步IO密集型任务
    print("end asyncio.sleep")

async def main():
    print("-----f1-----")
    t1 = time.time()
    try:
        # 直接给整个f1()设置超时
        await asyncio.wait_for(f1(), timeout=2)
    except asyncio.TimeoutError:
        print("执行超时")
    except Exception as e:
        print(f"其他错误: {e}")
    t2 = time.time()
    print(f"f1 cost {(t2 - t1):.2f} s")

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

关键改动说明 #

• 拆分CPU密集任务：把time.sleep(3)放到普通函数cpu_bound_task中，通过线程池执行，避免阻塞异步事件循环；
• 正确调用异步函数：f1作为异步函数，内部通过await执行线程池任务和异步IO任务，保证整个流程是异步可调度的；
• 直接对f1设置超时：用asyncio.wait_for包裹f1()，超时会触发TimeoutError，实现对整个函数执行过程的超时控制。

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