FastAPI中asyncio.wait_for未按预期触发超时问题求助

验证uvloop是否为诱因 #

临时切换回标准asyncio事件循环，测试超时是否恢复正常，以此验证你的猜想：

import asyncio
from fastapi import FastAPI

# 禁用uvloop，使用默认事件循环
asyncio.set_event_loop_policy(asyncio.DefaultEventLoopPolicy())
app = FastAPI()

重启服务后复现场景，如果超时正常触发，说明uvloop存在兼容性问题；若问题依旧，排除uvloop，转向其他排查方向。

常见问题修复方案 #

1. 排查事件循环阻塞 #

如果completion_event.wait()等待期间，存在同步阻塞代码（比如CPU密集计算、同步IO操作）占用事件循环，wait_for的超时定时器会被彻底卡住。检查：

• 其他协程是否调用了time.sleep()、同步数据库查询这类同步操作？
• 同步代码是否未用run_in_executor包裹？改用异步实现或线程池执行同步逻辑。

2. 升级uvloop版本 #

旧版uvloop对asyncio.wait_for与Event的组合可能存在bug，执行升级命令：

pip install --upgrade uvloop

3. 手动实现超时逻辑 #

若确认是uvloop问题且不想放弃性能，用asyncio.wait替代wait_for，兼容性更强：

async def wait_with_timeout(event, timeout):
    timeout_task = asyncio.create_task(asyncio.sleep(timeout))
    event_task = asyncio.create_task(event.wait())
    done, pending = await asyncio.wait(
        [timeout_task, event_task],
        return_when=asyncio.FIRST_COMPLETED
    )
    # 清理未完成的任务
    for task in pending:
        task.cancel()
    return event_task in done  # 返回事件是否在超时前触发

# 替换原代码中的wait_for调用
event_triggered = await wait_with_timeout(completion_event, 1.0)
if not event_triggered:
    logging.info("timeout")
    return SubmissionResult(post_id=post_id, language_check_pending=True)

4. 检查Event触发逻辑 #

有可能completion_event被后台协程提前触发，导致wait()直接返回，看起来像超时未生效。添加日志确认事件状态：

logging.info(f"Event状态（等待前）: {completion_event.is_set()}")
try:
    await wait_for(completion_event.wait(), 1.0)
    logging.info("事件在超时前触发")
except TimeoutError:
    logging.info("timeout")
    return SubmissionResult(post_id=post_id, language_check_pending=True)

总结 #

先通过切换事件循环定位uvloop是否为问题根源，再逐一排查事件循环阻塞、版本问题，最终可通过手动实现超时逻辑确保功能稳定。

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