适用场景
FastAPI 服务平时响应正常，但只要某个文件处理、旧版 SDK 调用或报表计算接口并发升高，健康检查、登录和其他无关接口也一起变慢。CPU、内存和数据库连接数未必异常，扩容 Uvicorn worker 后只能暂时缓解。
这类问题常见于 async def 路由中直接调用同步阻塞函数。本文给出一个可复现的最小示例，说明如何区分事件循环阻塞、线程池排队和纯 CPU 饱和，并给出可测试的修复方式。
现象描述
典型现象不是单个下游接口超时，而是同一进程内看似无关的异步接口一起出现延迟尖峰：
/report执行同步文件读取或第三方 SDK 请求时，/health的 P99 同步升高；- 应用进程 CPU 并未跑满，数据库和下游 HTTP 服务也没有对应的慢查询；
- 增加 worker 数量后故障阈值提高，但单个 worker 仍会周期性“冻结”；
- 日志时间戳之间出现空档，异步定时任务也延后执行。
下面的代码虽然语法合法，却会让当前 worker 的事件循环停住两秒：
import time
from fastapi import FastAPI
app = FastAPI()
@app.get("/report")
async def create_report() -> dict[str, str]:
# 错误示例：同步阻塞函数在事件循环线程中直接执行。
time.sleep(2)
return {"status": "ready"}
@app.get("/health")
async def health() -> dict[str, str]:
return {"status": "ok"}
async def 不会自动把函数体中的同步代码变成异步代码。协程只有运行到能够让出控制权的 await 时，事件循环才有机会调度其他请求；time.sleep()、同步数据库驱动、同步 HTTP SDK 和普通文件 I/O 都可能在这段时间独占事件循环线程。
可能原因
排查时重点搜索以下调用：
- 在
async def中使用time.sleep()，而不是await asyncio.sleep()； - 调用
requests、同步云厂商 SDK、同步数据库客户端或阻塞式消息客户端； - 直接执行大文件读写、压缩、图片处理或模板渲染；
- 在事件循环线程中做大 JSON 编解码、加密、正则回溯或批量数据转换；
- 误以为普通工具函数会被 FastAPI 自动放入线程池。
FastAPI 只会把它直接调用的普通 def 路由和普通 def 依赖放入线程池。async def 路由内部直接调用的普通工具函数仍在当前线程执行，这是定位时最容易忽略的边界。
排查思路
1. 先证明事件循环发生延迟
在应用中增加一个低开销的事件循环延迟探针。它比较计划唤醒时间和实际唤醒时间，只记录超过阈值的异常，避免在热点循环中刷日志：
import asyncio
import logging
import time
logger = logging.getLogger(__name__)
async def monitor_event_loop(
interval_seconds: float = 0.5,
warning_seconds: float = 0.2,
) -> None:
while True:
started = time.monotonic()
await asyncio.sleep(interval_seconds)
lag_seconds = time.monotonic() - started - interval_seconds
if lag_seconds >= warning_seconds:
logger.warning(
"事件循环延迟超过阈值",
extra={"event_loop_lag_ms": round(lag_seconds * 1000)},
)
event_loop_lag_ms 持续升高说明协程没有按时获得执行机会。它不能单独指出是哪段代码，但可以把“下游慢”与“进程内部调度被阻塞”分开。
2. 用对照请求确认影响范围
压测可疑接口时，同时以低并发持续请求只返回固定 JSON 的 /health。若可疑接口启动后 /health 也出现接近相同长度的停顿，阻塞事件循环的可能性很高。
# 终端一：持续观察健康检查耗时
while ($true) {
curl.exe -s -o NUL -w "%{time_total}`n" http://127.0.0.1:8000/health
Start-Sleep -Milliseconds 200
}
# 终端二：触发 10 个并发请求，需要预先安装 hey
hey.exe -n 20 -c 10 http://127.0.0.1:8000/report
测试应在隔离环境执行，不要直接对生产接口施压。观察总吞吐的同时，还要看健康检查延迟、事件循环延迟和线程数，避免只凭平均响应时间判断。
3. 区分 I/O 阻塞与 CPU 密集计算
同步 I/O 的特点是 CPU 使用率不高，但线程长期等待文件、网络或驱动返回；这类调用适合异步客户端或受控线程池。CPU 密集任务的特点是某个核心持续繁忙，把它简单移到线程中不一定提高吞吐，还可能受 GIL 和线程竞争影响，应考虑进程池、任务队列或独立计算服务。
修复方案
方案一：优先换成真正的异步接口
如果库提供异步方法，直接等待它，让事件循环在 I/O 等待期间继续服务其他请求：
import asyncio
from fastapi import FastAPI
app = FastAPI()
@app.get("/report")
async def create_report() -> dict[str, str]:
await asyncio.sleep(2)
return {"status": "ready"}
真实项目中应对应替换为异步数据库驱动、异步 HTTP 客户端或库提供的异步 SDK。不要用 await asyncio.sleep(0) 包裹阻塞函数，它只能在调用之前让出一次控制权，不能改变后续阻塞行为。
方案二：把暂时无法替换的同步 I/O 移出事件循环
Python 3.9 及以上可使用 asyncio.to_thread()。同时用信号量限制并发，防止请求洪峰把线程池和下游服务一起压垮：
import asyncio
import time
from fastapi import FastAPI, HTTPException
app = FastAPI()
blocking_slots = asyncio.Semaphore(8)
def generate_report() -> str:
time.sleep(2)
return "ready"
@app.get("/report")
async def create_report() -> dict[str, str]:
try:
async with asyncio.timeout(5):
async with blocking_slots:
status = await asyncio.to_thread(generate_report)
except TimeoutError as exc:
raise HTTPException(status_code=503, detail="报表生成超时") from exc
return {"status": status}
to_thread() 解决的是事件循环被同步 I/O 卡住的问题，不代表底层工作已经可取消。外层超时后，正在执行的线程通常仍会继续运行，所以同步函数本身也应配置网络、数据库或文件操作超时。信号量容量应根据下游承载能力和压测结果设置，不能简单等于最大请求并发。
若整个路由都是同步 I/O，也可以把路径函数声明为普通 def，由 FastAPI 在线程池中执行：
@app.get("/legacy-report")
def create_legacy_report() -> dict[str, str]:
return {"status": generate_report()}
方案三：把 CPU 密集任务移出 Web 请求
图片批处理、压缩、模型推理或大规模报表计算不宜长期占用 Web worker。请求只负责校验参数、创建任务并返回任务 ID，由独立任务队列或进程池执行；客户端轮询状态或通过回调接收结果。这样可以单独设置并发、超时、重试和资源限额，也不会让计算峰值拖慢健康检查。
验证示例
修复应包含回归测试，验证阻塞工作运行期间其他协程仍能获得调度：
import asyncio
import time
import pytest
def blocking_operation() -> str:
time.sleep(0.2)
return "done"
@pytest.mark.asyncio
async def test_blocking_operation_does_not_block_event_loop() -> None:
started = time.monotonic()
task = asyncio.create_task(asyncio.to_thread(blocking_operation))
await asyncio.sleep(0.02)
scheduling_delay = time.monotonic() - started
assert scheduling_delay < 0.1
assert await task == "done"
时间断言容易受共享 CI 主机抖动影响，因此阈值要明显大于正常调度耗时。更完整的集成测试应同时请求慢接口和健康检查，并断言健康检查不会等待慢任务结束。
预防措施
- 代码审查中检查
async def内的同步 SDK、文件操作、time.sleep()和重计算； - 为外部 I/O 设置连接、读取和总耗时边界，不只依赖 Web 层超时；
- 对线程池任务、后台队列和下游调用分别设置并发上限与拒绝策略；
- 监控事件循环延迟、接口 P95/P99、线程池排队时间、活动线程数和任务队列长度；
- 将 CPU 密集工作从 Web 进程拆出，并为任务设计幂等键、失败状态和有限重试；
- 压测时同时观察业务接口与轻量健康检查，验证故障是否会跨接口扩散。
FastAPI 官方并发文档明确区分了普通 def 路由、async def 路由和直接调用的工具函数；Python 官方文档则说明了 asyncio.to_thread() 适合把会阻塞事件循环的 I/O 函数放到独立线程。实施前应结合当前 Python、FastAPI 和底层客户端版本核对接口行为。
参考资料：
总结
FastAPI 全站延迟不一定来自数据库或网络，也可能是某个 async def 路由直接执行同步阻塞代码。先用事件循环延迟和健康检查对照请求证明问题，再按工作类型选择异步客户端、受控线程池或独立任务系统。修复的关键不是机械地增加 worker 或线程数，而是让阻塞工作离开事件循环，并为超时、并发和失败恢复建立清晰边界。
Discussion
评论