使用asyncio.wait_for装饰器设置异步函数超时无效的问题排查

你的超时装饰器逻辑本身没问题，但被装饰函数里的time.sleep(30)是同步阻塞调用，直接导致asyncio的超时机制完全失效——同步代码会霸占整个线程，让事件循环根本没机会执行超时检查逻辑，必须等30秒的sleep执行完才会继续。

核心原因 #

asyncio基于单线程事件循环模型运转，所有任务调度都依赖事件循环的正常执行。time.sleep()是同步阻塞函数，它会让当前线程彻底暂停，事件循环在这30秒内无法处理任何其他操作，包括asyncio.wait_for的超时检测，自然没法在5秒后中断任务。

修复方案 #

方案1：替换为异步休眠

把同步sleep换成asyncio提供的异步休眠await asyncio.sleep(30)，这样事件循环在休眠期间可以正常调度其他任务，超时机制就能生效：

import asyncio

def await_time_limit(time_limit):
    def Inner(func):
        async def wrapper(*args, **kwargs):
            return await asyncio.wait_for(func(*args, **kwargs), time_limit)
        return wrapper
    return Inner

@await_time_limit(5)
async def whatever():
    await asyncio.sleep(30)  # 替换为异步sleep，超时机制生效

方案2：同步代码线程化（针对无法异步化的场景）

如果业务逻辑必须调用同步阻塞代码（比如某些不支持异步的第三方库），可以用asyncio.to_thread()把同步代码放到单独线程执行，避免阻塞事件循环：

import asyncio
import time

def await_time_limit(time_limit):
    def Inner(func):
        async def wrapper(*args, **kwargs):
            return await asyncio.wait_for(func(*args, **kwargs), time_limit)
        return wrapper
    return Inner

@await_time_limit(5)
async def whatever():
    await asyncio.to_thread(time.sleep, 30)  # 同步代码放入线程，超时机制生效

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