FastAPI 全局超时中间件对同步路由失效的解决方案-Python教程-PHP中文网

FastAPI 全局超时中间件对同步路由失效的解决方案

FastAPI 的全局超时中间件无法中断阻塞式同步函数（如 time.sleep()），因为 Python 的同步执行不可抢占；必须将路由改为异步并使用 asyncio.to_thread() 或协程原生等待，才能实现真正的可取消超时控制。

fastapi 的全局超时中间件无法中断阻塞式同步函数（如 time.sleep()），因为 python 的同步执行不可抢占；必须将路由改为异步并使用 asyncio.to_thread() 或协程原生等待，才能实现真正的可取消超时控制。

在 FastAPI 中，TimeoutMiddleware 依赖 asyncio.wait_for() 实现请求超时，而该机制仅对 可等待对象（awaitables） 有效——即真正的异步协程或通过 asyncio.to_thread()/loop.run_in_executor() 提交的线程任务。当路由函数是同步函数（哪怕被声明为 async def 但内部调用 time.sleep() 等阻塞操作）时，事件循环会被完全阻塞，asyncio.TimeoutError 无法触发，中间件超时逻辑形同虚设。

✅ 正确做法：统一使用异步风格，并将同步阻塞操作移入线程池
以下为推荐的健壮实现：

from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request, HTTPException
from starlette.responses import JSONResponse
import asyncio
import time

class TimeoutMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, timeout: float = 5.0):
        super().__init__(app)
        self.timeout = timeout

    async def dispatch(self, request: Request, call_next):
        try:
            # ✅ wait_for 可取消：仅对协程/awaitable 生效
            response = await asyncio.wait_for(call_next(request), timeout=self.timeout)
            return response
        except asyncio.TimeoutError:
            return JSONResponse(
                status_code=408,
                content={"detail": "Request timeout exceeded"}
            )
        except Exception as e:
            raise e

# 示例：安全地执行同步阻塞代码（如 legacy DB calls、CPU-bound ops）
@app.get("/sync-heavy")
async def sync_heavy_route():
    # ❌ 错误：直接调用阻塞函数会冻结 event loop
    # time.sleep(6)  # → 超时中间件失效！

    # ✅ 正确：委托给线程池，保持协程可取消性
    await asyncio.to_thread(time.sleep, 6)  # Python 3.9+
    return {"status": "done"}

# 兼容旧版 Python（<p>⚠️ 关键注意事项：  </p><div class="aritcle_card flexRow artxards">
											<div class="artcardd flexRow">
												<a class="aritcle_card_img" rel="nofollow" href="/xiazai/skill7176" title="FastAPI Flask Proxy"><img
														src="https://img.php.cn/upload/skill/000/000/081/179134925252133.jpg" alt="FastAPI Flask Proxy" onerror="this.onerror='';this.src='/static/lhimages/moren/morentu.png'" ></a>
												<div class="aritcle_card_info flexColumn">
													<a rel="nofollow" href="/xiazai/skill7176" title="FastAPI Flask Proxy" class="overflowclass">FastAPI Flask Proxy</a>
													<p class="overflowclass">FastAPI + Flask 混合部署最佳实践，解决路由定义、API 代理等常见问题，适用于同时运行 FastAPI API 与 Flask 前端的场景。</p>
												</div>
												<a rel="nofollow" href="/xiazai/skill7176" title="FastAPI Flask Proxy" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span>
												</a>
											</div>
										</div>

async def + time.sleep() ≠ 异步：它仍是同步阻塞，不会让出控制权，必须显式使用 asyncio.to_thread() 包装；
不要试图用信号（signal.alarm）或多进程强制终止同步任务——FastAPI 运行在单进程多线程/协程模型下，信号不可靠且不跨平台；
对于 CPU 密集型同步逻辑（如图像处理、加密计算），优先考虑 asyncio.to_thread()；若需更高性能，应重构为真正异步库（如 httpx.AsyncClient 替代 requests）；
所有中间件超时均基于 协程执行时间，不包含响应体传输耗时（如大文件流式返回），如需端到端超时，需结合反向代理（如 Nginx proxy_read_timeout）。