FastAPI 的路由用 def 还是 async def？写错会怎样？

简化版

FastAPI 允许路由函数写成 def 或 async def，两者的执行方式完全不同：写 async def 的函数直接在事件循环里运行（不切线程，性能最好）；写 def 的函数 FastAPI 会自动丢到一个线程池里执行（run_in_threadpool，底层是 anyio 的线程池，默认 40 个线程），这样同步的阻塞代码就不会卡住事件循环。所以选择规则只有一条：函数里用的是异步库还是同步库——用 httpx.AsyncClient、asyncpg、AsyncSession 这类异步库就写 async def；用 requests、同步的 SQLAlchemy Session、time.sleep、PIL 这类同步/阻塞代码就写 def。最危险的错误是「在 async def 里调用阻塞代码」——比如 async def 里写 requests.get() 或 time.sleep(3)，它会霸占事件循环线程，导致整个进程的所有请求全部卡住（不是只有这一个请求慢，而是所有并发请求都被阻塞），这是 FastAPI 最典型的性能事故。反过来「该用 async def 却写了 def」只是损失一点性能（多一次线程切换，且受 40 线程上限约束），不会致命。必须在 async def 里执行阻塞代码时，用 await run_in_threadpool(fn, ...)（IO 阻塞）或 await loop.run_in_executor(ProcessPoolExecutor(), fn)（CPU 密集）。同样的规则也适用于依赖（Depends）和中间件——它们同样区分 def 和 async def。核心记忆：async def 直接跑在事件循环、def 丢线程池；用异步库写 async def，用同步库写 def；绝不在 async def 里写阻塞代码。

详细版

两种写法的执行路径对比：

	`async def`	`def`
执行位置	事件循环（主线程）	anyio 线程池
并发上限	理论上很高（受 IO 限制）	默认 40 个线程
切换开销	无	有（线程调度）
里面能用同步阻塞吗	绝对不能	可以（本来就是给它用的）
适用	`httpx`/`asyncpg`/`AsyncSession`	`requests`/同步 `Session`/`PIL`
写错的后果	阻塞整个进程	只是慢一点

from fastapi import FastAPI, Depends
from fastapi.concurrency import run_in_threadpool
import anyio, asyncio, time, httpx, requests

app = FastAPI()

# ① ★async def：用异步库（★推荐★）★
@app.get("/async-ok")
async def async_ok():
    async with httpx.AsyncClient() as client:        # ★异步 HTTP★
        r = await client.get("https://api.example.com", timeout=5)
    return r.json()

# ② ★def：用同步库（FastAPI 自动丢线程池）★
@app.get("/sync-ok")
def sync_ok():
    r = requests.get("https://api.example.com", timeout=5)   # ★阻塞但在线程里★
    return r.json()

# ③ ★★致命错误：async def 里写阻塞代码★★
@app.get("/async-bad")
async def async_bad():
    r = requests.get("https://api.example.com")      # ★★阻塞事件循环！★★
    time.sleep(3)                                     # ★★整个进程卡 3 秒★★
    return r.json()
# → ★这期间所有并发请求全部无响应★（不只是这一个）

# ④ ★必须在 async 里跑阻塞代码时★
@app.get("/mixed")
async def mixed():
    # ★IO 阻塞 → 线程池★
    data = await run_in_threadpool(requests.get, "https://x.com")
    # ★等价写法★
    data = await anyio.to_thread.run_sync(blocking_io_fn, arg)
    # ★CPU 密集 → 进程池（★线程池救不了 CPU 密集★）★
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(process_pool, cpu_heavy_fn, arg)
    return {"ok": True}

# ⑤ ★依赖也区分 def / async def（★同样规则★）★
def sync_dep(db: Session = Depends(get_db)):          # ★→ 线程池★
    return db.query(User).first()

async def async_dep(db: AsyncSession = Depends(get_async_db)):   # ★→ 事件循环★
    return (await db.execute(select(User))).scalar_one_or_none()

# ⑥ ★调整线程池大小★
@app.on_event("startup")     # 或 lifespan
async def set_threads():
    limiter = anyio.to_thread.current_default_thread_limiter()
    limiter.total_tokens = 100                        # ★★默认 40★★

# ⑦ ★检测阻塞（开发期利器）★
import asyncio
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.slow_callback_duration = 0.1          # ★★超过 100ms 的回调会警告★★
# 或用 pip install blockbuster / aiodebug

⚠️ 三个必须记住的点：① async def 里的阻塞调用会卡住整个进程，而不只是当前请求。事件循环是单线程的——它靠「遇到 await 就切到别的任务」来实现并发。一旦你在 async def 里执行 requests.get()（网络等待）、time.sleep(3)、同步的数据库查询、或者一个耗时的 CPU 计算，事件循环线程就被完全占住了，期间它无法处理任何其他请求：100 个并发请求会变成串行，P99 延迟直接爆炸。这是 FastAPI 项目最常见也最隐蔽的性能事故——因为功能完全正常，只有压测或线上流量上来才暴露。② def 路由不是「不好」，而是「有上限」。FastAPI 会用 run_in_threadpool 把它丢进 anyio 的默认线程池，这个线程池默认只有 40 个 token——也就是说同时最多 40 个 def 请求在跑，第 41 个开始排队。如果你的同步接口每个要 200ms，那这个应用的同步部分吞吐上限就是 40 / 0.2 = 200 QPS。可以调大 total_tokens，但线程不是免费的（每个线程约 8MB 栈空间，且 GIL 下 CPU 密集任务并不会真并行）。③ 同样的规则适用于依赖和中间件。Depends 的依赖函数如果是 def 会走线程池、async def 直接在循环里跑；BaseHTTPMiddleware 里的阻塞代码同样会卡死事件循环。特别要注意依赖里的同步数据库 session——那是最常见的「不知不觉写了阻塞代码」的地方。

完整版教学

一、事件循环为什么怕阻塞

★ 单线程事件循环的工作方式：
  ┌──────────────────────────────────────────────────────┐
  │ 事件循环（★一个线程★）                                  │
  │   任务A: 执行 → ★await（IO 等待）★ → ★挂起，切到 B★      │
  │   任务B: 执行 → ★await★ → 挂起，切到 C                  │
  │   任务C: 执行 → 完成                                    │
  │   任务A: IO 好了 → 恢复执行 → 完成                       │
  └──────────────────────────────────────────────────────┘
  ★ ★并发的本质：在等 IO 的时候去干别的★
  ★ ★前提：必须"主动交出控制权"（await）★

★ ★阻塞发生了什么★：
  任务A: 执行 → ★time.sleep(3) / requests.get()★
              ↓
         ★★线程被占住，不返回事件循环★★
              ↓
  任务B、C、D... ★全部无法执行★（哪怕它们的 IO 早就好了）
  ★ → ★一个请求拖垮全部★

★ ★量化对比（100 个并发请求，每个要等 100ms IO）★：
  ┌──────────────────────────────┬────────────────────┐
  │ ★async def + await（正确）★   │ ★总耗时 ≈ 100ms★    │
  │                               │ （100 个并发等待）  │
  ├──────────────────────────────┼────────────────────┤
  │ ★async def + 阻塞调用（错误）★ │ ★总耗时 ≈ 10 秒★    │
  │                               │ ★（完全串行！）★    │
  ├──────────────────────────────┼────────────────────┤
  │ ★def（线程池，40 线程）★       │ ★总耗时 ≈ 300ms★    │
  │                               │ （40 并发 × 3 批）  │
  └──────────────────────────────┴────────────────────┘
  ★ ★结论：写错 async 比用 def 慢 30 倍★

★ ★为什么 def 反而"安全"★：
  FastAPI 检测到路由函数不是协程 → ★自动 run_in_threadpool★
  → ★阻塞发生在工作线程里，事件循环照常运转★
  → ★这是 FastAPI 相比裸 asyncio 框架的贴心之处★
  ★ 源码大意（fastapi/routing.py）：
    if is_coroutine_callable(func):
        result = await func(**values)              # ★直接 await★
    else:
        result = await run_in_threadpool(func, **values)   # ★线程池★

★ ★GIL 的影响★：
  线程池能解决 ★IO 阻塞★（等待时 GIL 会释放）
  ★但解决不了 CPU 密集★（GIL 下多线程不能真并行）
  → ★CPU 密集必须用进程池或独立的 worker 服务★
  ★ 例：图片处理、大 JSON 解析、加密、模型推理

理解「为什么怕阻塞」要先理解事件循环的并发本质：在等 IO 的时候去干别的——前提是必须主动交出控制权（await）。一旦在 async def 里执行阻塞调用，线程被占住不返回事件循环，其他所有任务全部无法执行。量化对比很直观：100 个并发、每个等 100ms IO 的场景下，正确的 async def 总耗时约 100ms、写错的 async def 要 10 秒（完全串行）、而 def 走线程池约 300ms——写错 async 比老老实实用 def 慢 30 倍。FastAPI 的贴心之处在于它会检测函数是不是协程，不是的话自动 run_in_threadpool。但要注意 GIL 的边界：线程池能解决 IO 阻塞（等待时 GIL 会释放），解决不了 CPU 密集——那必须用进程池或独立的 worker 服务。

二、怎么判断该用哪个

★ ★唯一的判断依据：函数体里用的库是同步还是异步★

  ┌────────────────────────────┬──────────────────────────┐
  │ ★异步库 → async def★        │ ★同步库 → def★            │
  ├────────────────────────────┼──────────────────────────┤
  │ httpx.AsyncClient          │ requests                  │
  │ aiohttp                    │ urllib                    │
  │ asyncpg / aiomysql         │ psycopg2 / pymysql        │
  │ ★SQLAlchemy AsyncSession★  │ ★SQLAlchemy Session★      │
  │ motor（异步 MongoDB）       │ pymongo                   │
  │ redis.asyncio              │ redis（同步）              │
  │ aiofiles                   │ open()                    │
  │ asyncio.sleep              │ time.sleep                │
  └────────────────────────────┴──────────────────────────┘

★ ★纯计算的路由怎么选★：
  @app.get("/calc")
  def calc():                    # ★★如果计算 > 几十毫秒，用 def★★
      return heavy_math()        # → 丢线程池，不卡事件循环
  ★ 但 ★GIL 下线程池也不能真并行 CPU★
  → ★真正的 CPU 密集：进程池 或 拆成独立服务★

★ ★"什么都不做"的路由★：
  @app.get("/health")
  async def health():            # ★★用 async def（零开销）★★
      return {"status": "ok"}
  ★ 不涉及任何 IO/计算 → async def 最快（★省一次线程切换★）

★ ★混合场景（★最常见的现实问题★）★：
  一个接口既要查异步数据库，又要调一个只有同步 SDK 的第三方
  ✓ 写 async def，把同步部分丢线程池：
    @app.get("/mixed")
    async def mixed():
        user = await db.get(User, uid)                    # 异步
        result = await run_in_threadpool(sync_sdk.call, x) # ★同步丢线程★
        return {...}
  ✗ 写 def 然后在里面 asyncio.run(...)   # ★★绝对不行★★
    → ★在已有事件循环的线程里再起一个循环会报错★
    → 在线程池的线程里 asyncio.run 虽然不报错，但★白白浪费★

★ ★判断流程图★：
  这个路由函数里有没有 IO？
   ├─ 没有（纯计算/纯内存）
   │   ├─ 很快（<10ms） → ★async def★
   │   └─ 很慢 → ★def★（或丢进程池）
   └─ 有 IO
       ├─ ★用的是异步库★ → ★async def + await★
       ├─ ★用的是同步库★ → ★def★
       └─ 混合 → ★async def + run_in_threadpool 包同步部分★

★ ★"我全写 async def 不行吗"★：
  ✗ ★除非你确认里面没有任何同步阻塞★
  ★ 常见的隐蔽阻塞：
    - ORM 的 ★懒加载★（访问一个未加载的关系会发同步 SQL）
    - ★日志写文件★（通常很快，可忽略）
    - ★读配置文件、open()★
    - ★第三方 SDK 内部的 requests★
    - ★DNS 解析（socket.getaddrinfo）★
    - ★加密/哈希（bcrypt 故意很慢）★  ← ★★登录接口的经典坑★★

判断依据只有一条：函数体里用的库是同步还是异步——那张对照表要记住（httpx.AsyncClient vs requests、AsyncSession vs Session、asyncio.sleep vs time.sleep）。纯计算的路由如果超过几十毫秒就该用 def（丢线程池不卡事件循环），但 GIL 下线程池也不能真并行 CPU，真正的 CPU 密集要用进程池或独立服务。「什么都不做」的路由（健康检查）用 async def 最快（省一次线程切换）。混合场景是最常见的现实问题——写 async def 然后把同步部分用 run_in_threadpool 包起来；绝不能在 def 里 asyncio.run(...)。至于「全写 async def 行不行」——除非你确认里面没有任何同步阻塞，而常见的隐蔽阻塞包括 ORM 懒加载、第三方 SDK 内部的 requests、DNS 解析、以及 bcrypt 这类故意很慢的密码哈希（登录接口的经典坑）。

三、线程池的机制与调优

★ ★FastAPI 用的是 anyio 的默认线程池★：
  from fastapi.concurrency import run_in_threadpool
  # 内部：await anyio.to_thread.run_sync(func, *args)
  ★ ★默认容量：40 个 token★（anyio 的 total_tokens）

★ ★40 意味着什么（算例）★：
  同步接口平均耗时 200ms
  → 单进程同步部分吞吐 = ★40 / 0.2 = 200 QPS★
  → 第 41 个并发请求开始★排队等待★
  → ★表现：QPS 上不去、P99 陡增，但 CPU 很闲★
  ★ ★这个"很闲却慢"的现象是线程池打满的典型特征★

★ ★调整线程池★：
  from contextlib import asynccontextmanager
  import anyio.to_thread

  @asynccontextmanager
  async def lifespan(app):
      limiter = anyio.to_thread.current_default_thread_limiter()
      limiter.total_tokens = 100                # ★调大★
      yield
  app = FastAPI(lifespan=lifespan)

  ★ ★调多大合适★：
    - IO 等待为主 → ★可以调到 100~200★（线程大部分时间在睡）
    - ★但要考虑：★
      ① ★每个线程约 8MB 栈★ → 200 线程 ≈ 1.6GB 虚拟内存
      ② ★数据库连接数★：线程数 × worker 数 ≤ 数据库上限
      ③ ★线程切换开销★（几百个线程时开始明显）
    - ★CPU 密集 → 调大没用★（GIL）

★ ★另一个思路：多 worker★：
  uvicorn app:app --workers 4
  → ★4 个独立进程，各有自己的事件循环和线程池★
  → 总线程池容量 = ★4 × 40 = 160★
  ★ ★进程数 × 线程数 = 数据库连接的需求上限★（容易爆）

★ ★怎么发现线程池被打满★：
  ① ★现象：CPU 空闲但响应慢、QPS 有天花板★
  ② 监控：
     limiter = anyio.to_thread.current_default_thread_limiter()
     limiter.available_tokens          # ★剩余可用★
     limiter.statistics()
  ③ 定期上报：
     @app.middleware("http")
     async def track(request, call_next):
         lim = anyio.to_thread.current_default_thread_limiter()
         metrics.gauge("threadpool.available", lim.available_tokens)
         return await call_next(request)

★ ★CPU 密集：进程池★：
  from concurrent.futures import ProcessPoolExecutor
  pool = ProcessPoolExecutor(max_workers=4)

  @app.get("/heavy")
  async def heavy():
      loop = asyncio.get_running_loop()
      result = await loop.run_in_executor(pool, cpu_bound_fn, arg)
      return result
  ★ 注意：
    ① ★参数和返回值要能 pickle★
    ② ★进程创建开销大★ → 复用 pool，别每次新建
    ③ ★不能传数据库连接、文件句柄★
    ④ ★更好的方案：拆成独立的 worker 服务（Celery/RQ）★

FastAPI 用的是 anyio 的默认线程池，容量是 40 个 token。这个数字很关键：同步接口平均 200ms 的话，单进程同步部分的吞吐上限就是 200 QPS——表现为「QPS 上不去、P99 陡增，但 CPU 很闲」，这个「很闲却慢」的现象是线程池打满的典型特征。调大 total_tokens 时要考虑三点：每个线程约 8MB 栈空间、线程数 × worker 数不能超过数据库连接上限、线程切换开销；而且 CPU 密集调大没用（GIL）。可以在中间件里上报 limiter.available_tokens 做监控。CPU 密集要用进程池（run_in_executor），注意参数要能 pickle、要复用 pool、更好的方案其实是拆成独立的 worker 服务。

四、隐蔽的阻塞点

★ ★★① 密码哈希（登录接口的经典事故）★★：
  @app.post("/login")
  async def login(data: LoginIn):
      if bcrypt.checkpw(data.password.encode(), user.hash):   # ★★~100~300ms★★
          ...
  ★ ★bcrypt/scrypt/argon2 是故意设计成慢的★（防暴力破解）
  → ★在 async def 里 = 每次登录卡住事件循环 200ms★
  → ★登录高峰时整站无响应★
  ✓ 改成 def 路由，或：
    valid = await run_in_threadpool(bcrypt.checkpw, pwd, hashed)

★ ★② ORM 懒加载★：
  @app.get("/users")
  async def users(db: AsyncSession = Depends(get_db)):
      users = (await db.execute(select(User))).scalars().all()
      return [{"name": u.name, "posts": len(u.posts)} for u in users]
                                              # ★★u.posts 触发同步 IO！★★
  → ★async SQLAlchemy 下会直接抛 MissingGreenlet 异常★
  ✓ ★预加载★：select(User).options(selectinload(User.posts))

★ ★③ 第三方 SDK★：
  import boto3, stripe                # ★内部都是 requests★
  @app.post("/pay")
  async def pay():
      stripe.Charge.create(...)        # ★★阻塞★★
  ✓ await run_in_threadpool(stripe.Charge.create, ...)
  ✓ 或用官方的异步版本（aioboto3 等）

★ ★④ 同步 Redis / 缓存客户端★：
  import redis                        # ★同步★
  r = redis.Redis()
  @app.get("/x")
  async def x():
      return r.get("key")             # ★★阻塞★★
  ✓ import redis.asyncio as aioredis  # ★异步版★

★ ★⑤ 文件读写★：
  async def upload(f: UploadFile):
      content = await f.read()        # ✓ ★UploadFile 的方法是异步的★
      with open(path, "wb") as out:   # ★★同步写，大文件会阻塞★★
          out.write(content)
  ✓ await run_in_threadpool(save_file, path, content)
  ✓ 或 aiofiles

★ ★⑥ 大 JSON 的序列化/解析★：
  返回 10MB 的 JSON → ★json.dumps 可能耗时几百毫秒★
  ✓ ★用 orjson（快 5~10 倍）★
  ✓ ★分页★（根本上减少数据量）

★ ★⑦ DNS 解析★：
  socket.getaddrinfo 是同步的
  → ★第一次连接新域名时可能阻塞几十毫秒★
  ★ aiohttp/httpx 内部有处理，但自己写 socket 要注意

★ ★⑧ 中间件里的阻塞★：
  @app.middleware("http")
  async def log_mw(request, call_next):
      log_to_db(request)               # ★★同步写库，每个请求都阻塞★★
      return await call_next(request)
  ★ ★中间件的阻塞影响所有请求★，比单个路由更严重

★ ★怎么系统地发现★：
  ① ★asyncio debug 模式★：
     loop.set_debug(True); loop.slow_callback_duration = 0.1
     → ★"Executing <Task...> took 0.523 seconds" 警告★
  ② ★blockbuster 库★（自动检测阻塞调用）
  ③ ★压测对比★：并发 1 和并发 50 的 P99 差多少
     → ★如果并发上去后 P99 线性恶化 → 大概率有阻塞★
  ④ py-spy dump --pid X    # ★看线程栈都卡在哪★

隐蔽阻塞点里最经典的是密码哈希——bcrypt/argon2 故意设计成慢的（100300ms 防暴力破解），在 async def 里就是每次登录卡住事件循环 200ms，登录高峰时整站无响应。其他七个：ORM 懒加载（async SQLAlchemy 下会直接抛 MissingGreenlet）、第三方 SDK（boto3/stripe 内部都是 requests）、同步 Redis 客户端、文件同步写、大 JSON 序列化（用 orjson 快 510 倍）、DNS 解析、以及中间件里的阻塞（影响所有请求，比单个路由更严重）。系统发现的方法：asyncio debug 模式（slow_callback_duration = 0.1 会打印「took 0.523 seconds」警告）、blockbuster 库、以及压测对比——如果并发上去后 P99 线性恶化，大概率有阻塞。

五、依赖、中间件与后台任务

★ ★依赖（Depends）同样区分★：
  def get_db():                        # ★→ 线程池★
      db = SessionLocal()
      try: yield db
      finally: db.close()

  async def get_async_db():            # ★→ 事件循环★
      async with AsyncSessionLocal() as session:
          yield session

  ★ ★混用的后果★：
    async def 路由 + def 依赖 → ★依赖走线程池，路由在循环里★（★可以，正常★）
    def 路由 + async def 依赖 → ★依赖在循环里，路由在线程池★（★也可以★）
  ★ FastAPI 会正确处理，但★同步依赖也占用同一个 40 线程池★
    → ★依赖 + 路由都是 def 时，一个请求可能占 2 个 token？★
      实际上是顺序执行，但★依赖里的阻塞同样计入线程池压力★

★ ★yield 依赖的清理时机★：
  async def get_db():
      async with AsyncSessionLocal() as s:
          yield s                      # ★★请求处理完后才执行退出逻辑★★
  ★ 注意：★yield 之后的代码在响应发送之后执行★（0.106+ 的行为）
  → ★不要在这里做耗时操作★（会延长请求占用）

★ ★中间件的两种写法★：
  # ① ★@app.middleware("http")（基于 BaseHTTPMiddleware）★
  @app.middleware("http")
  async def add_header(request, call_next):
      resp = await call_next(request)
      resp.headers["X-Time"] = "..."
      return resp
  ★ ✗ 已知问题：★BaseHTTPMiddleware 会破坏 StreamingResponse 的流式★
    ✗ ★对 BackgroundTasks 的时序也有影响★
    ✗ 性能开销比纯 ASGI 中间件大

  # ② ★纯 ASGI 中间件（★性能更好、无副作用★）★
  class TimingMiddleware:
      def __init__(self, app): self.app = app
      async def __call__(self, scope, receive, send):
          if scope["type"] != "http":
              return await self.app(scope, receive, send)
          start = time.perf_counter()
          async def send_wrapper(message):
              if message["type"] == "http.response.start":
                  dur = (time.perf_counter() - start) * 1000
                  message["headers"].append(
                      (b"x-process-time", f"{dur:.1f}".encode()))
              await send(message)
          await self.app(scope, receive, send_wrapper)
  app.add_middleware(TimingMiddleware)
  ★ ★需要流式响应/高性能时用这种★

★ ★BackgroundTasks 同样区分★：
  @app.post("/send")
  async def send(bt: BackgroundTasks):
      bt.add_task(sync_send_email, addr)    # ★def → 线程池★
      bt.add_task(async_send_email, addr)   # ★async def → 事件循环★
      return {"ok": True}
  ★ ★BackgroundTasks 在响应发送后执行，但仍在同一个进程里★
  → ★耗时任务会占用 worker 资源★ → ★重活交给 Celery★

★ ★WebSocket 里必须是 async★：
  @app.websocket("/ws")
  async def ws(websocket: WebSocket):      # ★★只能 async def★★
      await websocket.accept()
      while True:
          data = await websocket.receive_text()
          # ★这里的阻塞会影响所有连接★

依赖同样区分 def 和 async def——def 依赖走线程池、async def 依赖在事件循环里，FastAPI 会正确处理混用，但同步依赖也会占用同一个 40 线程池。要注意 yield 依赖的清理代码在响应发送之后执行，所以不要在那里做耗时操作。中间件有两种写法：@app.middleware("http") 基于 BaseHTTPMiddleware，已知会破坏 StreamingResponse 的流式效果、影响 BackgroundTasks 时序、性能开销也更大；需要流式响应或高性能时应该写纯 ASGI 中间件（直接操作 scope/receive/send）。BackgroundTasks 也区分两种函数，而且它在响应发送后执行但仍在同一进程里——耗时任务会占用 worker 资源，重活要交给 Celery。WebSocket 处理函数只能是 async def，里面的阻塞会影响所有连接。

六、实践清单

★ 决策速查：
  ┌────────────────────────────────┬──────────────────┐
  │ 用 httpx.AsyncClient / asyncpg  │ ★async def★       │
  │ 用 requests / 同步 Session      │ ★def★             │
  │ 纯计算 < 10ms（健康检查）        │ ★async def★       │
  │ 纯计算 > 50ms                   │ ★def 或进程池★    │
  │ ★密码哈希（bcrypt）★            │ ★def 或线程池★    │
  │ 混合（异步 DB + 同步 SDK）       │ ★async def +      │
  │                                 │ run_in_threadpool★│
  │ WebSocket                       │ ★只能 async def★  │
  └────────────────────────────────┴──────────────────┘

★ 检查清单：
  □ ★async def 里没有 requests / time.sleep / 同步 DB★
  □ ★async def 里没有 bcrypt / 大文件读写 / 重计算★
  □ ★第三方 SDK 确认过是同步还是异步★
  □ ★async ORM 用了预加载（避免懒加载触发 IO）★
  □ ★中间件里没有阻塞调用★
  □ ★线程池容量评估过（默认 40）★
  □ ★CPU 密集用进程池或独立服务★
  □ ★开发环境开 asyncio debug 检测慢回调★
  □ ★压测验证并发上去后 P99 不恶化★

★ ★排查"并发上不去"的顺序★：
  ① ★CPU 是不是满的★
     - ★满 → CPU 密集，加 worker 或拆服务★
     - ★不满 → 继续往下查（多半是阻塞或线程池打满）★
  ② ★看 anyio 线程池剩余 token★
  ③ ★asyncio debug 看有没有慢回调警告★
  ④ ★py-spy dump 看线程栈卡在哪★
  ⑤ ★检查数据库连接池是否耗尽★
  ⑥ 检查下游服务是否变慢（有没有配超时）

★ ★一个真实的排查故事（典型）★：
  现象：QPS 只有 30，CPU 占用 15%，内存正常
  排查：py-spy 显示多数线程卡在 ★bcrypt.checkpw★
  原因：★登录接口写了 async def，bcrypt 阻塞事件循环★
  修复：★改成 def 路由★ → QPS 升到 400+
  ★ ★教训：CPU 闲但慢 = 阻塞或池打满★

★ 一句话总结：
  ★"async def 直接跑在事件循环、def 自动丢进 40 个线程的线程池；
    判断依据只有一条——函数里用的是异步库还是同步库；
    最致命的错误是在 async def 里写阻塞代码（卡死整个进程，
    比用 def 还慢 30 倍）；必须混用时用 run_in_threadpool 包起来。"★

检查清单里最容易漏的三条：第三方 SDK 要确认是同步还是异步（boto3、stripe 都是同步的）、async ORM 要用预加载避免懒加载触发 IO、中间件里不能有阻塞调用。排查「并发上不去」有固定顺序：先看 CPU 满不满——满了是 CPU 密集（加 worker 或拆服务），不满则多半是阻塞或线程池打满；然后看 anyio 剩余 token、asyncio debug 的慢回调警告、py-spy dump 看线程栈。那个真实排查故事很典型：QPS 只有 30、CPU 占用 15%，最后发现是登录接口写了 async def 而 bcrypt 阻塞了事件循环，改成 def 后 QPS 升到 400+。

记忆钩子：「FastAPI 的路由可以写 def 或 async def，★两者执行路径完全不同★：★async def 直接在事件循环里跑★（不切线程，性能最好）；★def 会被 FastAPI 自动丢进 anyio 的线程池（run_in_threadpool），默认只有 40 个 token★。★选择依据只有一条：函数体里用的是异步库还是同步库★——httpx.AsyncClient/asyncpg/AsyncSession/redis.asyncio/aiofiles 用 async def，requests/同步 Session/pymongo/time.sleep/open() 用 def。★最致命的错误是在 async def 里写阻塞代码★——事件循环是★单线程★的，靠『遇到 await 就切到别的任务』实现并发，一旦被 requests.get() 或 time.sleep(3) 占住，★整个进程的所有并发请求全部卡住★（不是只有这一个慢）。★量化：100 个并发各等 100ms IO，正确的 async def 约 100ms、写错的 async def 要 10 秒（完全串行）、def 走线程池约 300ms —— 写错 async 比老实用 def 慢 30 倍★。反过来『该 async 却写了 def』只是慢一点，★但受 40 线程上限约束★：同步接口 200ms 时单进程吞吐上限就是 40/0.2=200 QPS，★典型特征是『CPU 很闲但 QPS 上不去、P99 陡增』★，可以在 lifespan 里调 ★anyio.to_thread.current_default_thread_limiter().total_tokens★（但每线程约 8MB 栈、且 ★GIL 下调大对 CPU 密集无用★）。必须混用时：★IO 阻塞用 await run_in_threadpool(fn, …)、CPU 密集用 loop.run_in_executor(ProcessPoolExecutor(), fn)★，★绝不能在 def 里 asyncio.run()★。★八个隐蔽阻塞点★：★① bcrypt/argon2 密码哈希（故意设计成慢的 100~300ms，登录接口的经典事故）★ ② ★ORM 懒加载（async SQLAlchemy 下直接抛 MissingGreenlet，要 selectinload 预加载）★ ③ ★第三方 SDK（boto3/stripe 内部是 requests）★ ④ 同步 redis 客户端 ⑤ 文件同步写 ⑥ 大 JSON 序列化（用 orjson）⑦ DNS 解析 ⑧ ★中间件里的阻塞（影响所有请求，比单个路由更严重）★。★依赖 Depends 和 BackgroundTasks 同样区分 def/async def★，★WebSocket 只能 async def★。★BaseHTTPMiddleware（@app.middleware(‘http’)）会破坏 StreamingResponse 的流式★，需要流式或高性能就写纯 ASGI 中间件。发现阻塞：★asyncio debug 模式 + slow_callback_duration=0.1★、blockbuster 库、★压测看并发上去后 P99 是否线性恶化★、py-spy dump 看线程栈。★排查口诀：CPU 闲但慢 = 阻塞或线程池打满★。」

七、常见误区与追问

误区：FastAPI 是异步框架，路由应该全部写成 async def。 这是最危险的误解。async def 的性能优势建立在「函数体内所有 IO 都是 await 的」这个前提上——如果里面用了 requests、同步的 SQLAlchemy Session、time.sleep()、或者任何一个内部用同步 IO 的第三方 SDK，那段代码会完全占住事件循环线程。事件循环是单线程的，它靠「任务主动交出控制权」来实现并发；一个不交出控制权的阻塞调用会让所有其他请求全部停摆——100 个并发请求从「同时等 100ms」变成「排队串行 10 秒」。反而是老老实实写 def更安全：FastAPI 检测到它不是协程函数，会自动 run_in_threadpool 丢进线程池，阻塞发生在工作线程里，事件循环照常运转。所以规则是看库不看框架：用异步库才写 async def。
误区：写成 def 会让 FastAPI 退化成同步框架，性能很差。 不会退化，只是有并发上限。FastAPI 会把 def 路由丢进 anyio 的默认线程池（run_in_threadpool），事件循环本身完全不受影响、依然能高效处理其他 async def 请求。真正的限制是线程池默认只有 40 个 token——同时最多 40 个 def 请求在执行，第 41 个开始排队。所以吞吐上限是 40 ÷ 单请求耗时：200ms 的接口就是 200 QPS。典型的症状是「CPU 很闲但 QPS 上不去、P99 陡增」。可以在 lifespan 里调大 anyio.to_thread.current_default_thread_limiter().total_tokens，但要权衡：每个线程约 8MB 栈空间、线程数 × worker 数不能超过数据库连接上限、几百个线程后切换开销开始明显；而且对 CPU 密集型任务调大线程数完全没用（GIL 让它们无法真正并行）。
误区：登录接口写 async def 没问题，验证密码很快。 密码哈希是故意设计成慢的——bcrypt、scrypt、argon2 的核心安全属性就是「计算代价高」，用来抵抗暴力破解，典型的 checkpw 耗时在 100~300 毫秒（work factor 越高越慢）。在 async def 里直接调用它，意味着每一次登录验证都把事件循环卡住两三百毫秒——平时看不出来，一旦遇到登录高峰（活动开始、上班打卡），几十个并发登录就能让整站所有接口都无响应。这是一个非常经典的生产事故，而且排查时很迷惑：CPU 占用不高、数据库很闲、日志也没有错误。修复方法有两个：把登录路由改成 def（最简单），或者保持 async def 但用 await run_in_threadpool(bcrypt.checkpw, ...) 把哈希计算丢到线程池。同类需要警惕的还有：图片处理、大文件的哈希计算、正则回溯、以及任何「故意慢」的加密操作。
误区：只要路由函数写对了就没问题，依赖和中间件不用管。 依赖和中间件同样会阻塞事件循环，而且中间件的影响更大。Depends 的依赖函数也区分 def 和 async def——写 def 会走线程池、写 async def 直接在循环里执行，所以在 async def 依赖里做同步数据库查询同样会卡死事件循环（这是最常见的「不知不觉写了阻塞代码」的地方，因为 get_db 之类的依赖往往是从教程里抄来的）。而中间件的阻塞更严重：@app.middleware("http") 注册的函数对每一个请求都会执行，在里面同步写数据库日志、同步调用鉴权服务、或者做重的序列化，等于给所有请求都加上了阻塞。此外 BackgroundTasks 的任务也区分两种函数，而且它虽然在响应发送后执行，但仍然占用同一个进程的资源——真正耗时的活应该交给 Celery 这类外部队列。
误区：@app.middleware("http") 是官方推荐写法，用它就对了。 它是最方便的写法，但基于 BaseHTTPMiddleware，有几个已知的副作用。① 破坏流式响应——BaseHTTPMiddleware 会把响应体收集起来再转发，导致 StreamingResponse 失去流式效果（SSE 推送变成一次性返回、大文件下载先在内存里攒齐），这是做流式接口时的经典踩坑。② 影响 BackgroundTasks 的执行时序。③ 性能开销更大——它内部用了额外的任务和内存流来桥接 ASGI 接口。④ 异常传播行为和纯 ASGI 中间件不完全一致。所以：简单的加响应头、记录耗时用它没问题；一旦应用里有流式响应（SSE、大文件、LLM 流式输出）或对性能敏感，就该写纯 ASGI 中间件——直接实现 async def __call__(self, scope, receive, send)，用 send_wrapper 包装 send 来修改响应，不会缓冲响应体。
追问：怎么系统地发现代码里的阻塞调用？ 四种手段，按投入排序。① 开发环境开 asyncio debug 模式——loop.set_debug(True) 加 loop.slow_callback_duration = 0.1，之后任何执行超过 100ms 没有 await 的回调都会打印警告「Executing <Task…> took 0.523 seconds」，这是零成本且最直接的方式（uvicorn 可以用 --loop asyncio 配合环境变量 PYTHONASYNCIODEBUG=1）。② 用专门的检测库——blockbuster 之类的工具会 monkey-patch 常见的阻塞函数（time.sleep、socket.recv、open），在事件循环线程里被调用时直接抛异常，能在测试阶段就把问题挡住。③ 压测对比——分别用并发 1 和并发 50 压同一个接口，如果 P99 随并发线性恶化（并发 50 时延迟正好是 50 倍），几乎可以断定有阻塞；正常的异步接口在并发上升时延迟应该基本平稳。④ 生产环境用 py-spy dump --pid X 看所有线程当前的调用栈——如果大量线程卡在同一个函数（比如 bcrypt.checkpw 或 socket.recv），凶手就找到了。
追问：run_in_threadpool 和 run_in_executor 有什么区别？ run_in_threadpool（FastAPI/Starlette 提供，底层是 anyio.to_thread.run_sync）用的是「全局共享的线程池」——就是那个默认 40 token 的池子，def 路由和 def 依赖也都在用它。好处是开箱即用、和 FastAPI 的并发控制一致；代价是你的手动调用会和框架自动调度的 def 路由抢同一批 token。loop.run_in_executor(executor, fn, *args) 是标准库 asyncio 的接口，可以传入你自己创建的 executor：传 ThreadPoolExecutor 就是独立的线程池（可以为特定用途隔离出专用的池，避免互相影响），传 ProcessPoolExecutor 就是进程池——这是处理 CPU 密集任务的唯一正确选择，因为线程池受 GIL 限制无法真正并行。使用进程池要注意三点：参数和返回值必须能 pickle（数据库连接、文件句柄、lambda 都不行）、进程创建开销大所以要复用同一个 pool、以及大对象的跨进程传输本身有序列化成本。当 CPU 密集任务比较重时，更好的架构其实是拆成独立的 worker 服务（Celery/RQ/独立的推理服务），而不是在 Web 进程里开进程池。
追问：多 worker 和调大线程池，该选哪个？ 两者解决的问题不同，通常要一起用。多 worker（uvicorn --workers 4 或 gunicorn 管理）是「多进程」——每个 worker 是独立的 Python 进程，有自己的事件循环、自己的 40 线程池、自己的内存空间。它的核心价值是绕过 GIL 利用多核 CPU：4 个 worker 能真正并行地执行 Python 字节码。调大线程池是「单进程内的并发度」——只对等待型的阻塞有效（等数据库、等 HTTP 响应），因为线程等待 IO 时会释放 GIL；对 CPU 密集完全无效。选择逻辑：CPU 使用率接近饱和 → 加 worker（或加机器）；CPU 很闲但 QPS 上不去 → 大概率是线程池打满或有阻塞，先查阻塞，确认无误再调大线程池。但两者都要受数据库连接数的约束——总连接需求 ≈ worker 数 × 每进程的连接池上限，8 个 worker 配 pool_size 10 就是 80 个连接，很容易超过数据库的 max_connections。经验配置是 worker 数 ≈ CPU 核数（IO 密集可以略多），线程池按实测的阻塞接口耗时来定。

八、加强记忆

FastAPI 的路由可以写 def 或 async def，两者执行路径完全不同：async def 直接在事件循环里跑（不切线程，性能最好）；def 会被 FastAPI 自动丢进 anyio 的线程池（run_in_threadpool），默认只有 40 个 token。选择依据只有一条：函数体里用的是异步库还是同步库——httpx.AsyncClient/asyncpg/AsyncSession/redis.asyncio/aiofiles 用 async def，requests/同步 Session/pymongo/time.sleep/open() 用 def。最致命的错误是在 async def 里写阻塞代码——事件循环是单线程的，靠「遇到 await 就切到别的任务」实现并发，一旦被 requests.get() 或 time.sleep(3) 占住，整个进程的所有并发请求全部卡住（不是只有这一个慢）。量化对比：100 个并发各等 100ms IO，正确的 async def 约 100ms、写错的 async def 要 10 秒（完全串行）、def 走线程池约 300ms——写错 async 比老实用 def 慢 30 倍。反过来「该用 async def 却写了 def」只是慢一点，但受 40 线程上限约束：同步接口 200ms 时单进程吞吐上限就是 40/0.2 = 200 QPS，典型特征是「CPU 很闲但 QPS 上不去、P99 陡增」；可以在 lifespan 里调 anyio.to_thread.current_default_thread_limiter().total_tokens，但每个线程约 8MB 栈，而且 GIL 下调大对 CPU 密集完全无用。必须混用时：IO 阻塞用 await run_in_threadpool(fn, ...)、CPU 密集用 loop.run_in_executor(ProcessPoolExecutor(), fn)，绝不能在 def 里 asyncio.run()。八个隐蔽的阻塞点：① bcrypt/argon2 密码哈希（故意设计成慢的 100~300ms，登录接口的经典事故）、② ORM 懒加载（async SQLAlchemy 下直接抛 MissingGreenlet，要用 selectinload 预加载）、③ 第三方 SDK（boto3/stripe 内部是 requests）、④ 同步 redis 客户端、⑤ 文件同步写、⑥ 大 JSON 序列化（用 orjson）、⑦ DNS 解析、⑧ 中间件里的阻塞（影响所有请求，比单个路由更严重）。依赖 Depends 和 BackgroundTasks 同样区分 def/async def，WebSocket 处理函数只能是 async def。还要知道 BaseHTTPMiddleware（即 @app.middleware("http")）会破坏 StreamingResponse 的流式效果，需要流式或高性能时应该写纯 ASGI 中间件。发现阻塞的手段：asyncio debug 模式 + slow_callback_duration = 0.1、blockbuster 库、压测看并发上去后 P99 是否线性恶化、py-spy dump 看线程栈。排查口诀：CPU 闲但慢 = 阻塞或线程池打满。

简化版

详细版