03 同步与异步 —— 你的 handler，到底该用 async def 还是 def？

关键词: async def, def, 事件循环, aiomysql, run_in_threadpool
难度: ★★★☆☆

场景导入：一个选择，牵动全局性能

图书管理 API 接入 MySQL 之后，每个 /books/{book_id} 请求都要经历一次真实的网络往返。到了第 13 章引入 create_async_engine 和 AsyncSessionLocal 以后，数据库相关的 handler 很自然地就该写成 async def，这样才能把异步栈的并发能力发挥出来。

但 FastAPI 的神奇之处在于，它并不强求你处处都用 async def。在某些场景下——比如纯 CPU 计算，或者被迫要调用一个同步阻塞库——老老实实用 def 反而更安全。

所以今天咱们就来彻底聊聊这件事：async def 和 def 到底该怎么选？答案其实比你想的简单：看函数体里有没有能 await 的 I/O。有一条就够，不用纠结其他的。

原理解析：框架在背后做了什么？

FastAPI 在判断怎么执行你的 handler 时，逻辑非常干净。它依赖 Starlette 里的 run_endpoint_function：如果你的函数是协程，就直接 await 它；如果是个普通函数，就把它交给 starlette.concurrency.run_in_threadpool，底层调用 anyio.to_thread.run_sync，丢到线程池里去跑。

顺着这个逻辑，判断准则可以浓缩成三条，日常开发完全够用了：

用 async def：函数体里的 I/O 操作全都有异步版本（网络请求、文件读写、aiomysql 这类异步数据库驱动），那就放心用。事件循环可以在单次请求内并发调度多个 await，吞吐上限取决于 I/O 并发数而非 CPU。咱们整个系列的默认选择就是这个。
用 def：你需要调一个同步阻塞库，而且它没有异步版本——比如 requests、某些 PDF 生成 SDK 或者加密库。这时候把整个 handler 定义成 def，Starlette 会自动把它投递到线程池里去执行，主事件循环完全不受影响。
纯 CPU 计算也用 def：比如哈希、序列化、图像处理。它和上一条走的是同一条线程池路径，框架替你兜底。但必须提醒一句：Python 的 GIL（全局解释器锁）会让多线程在 CPU 密集型任务上使不上劲，真要处理大量计算，还是得考虑进程池或者独立的任务系统。

这里还有一个特别实用的“逃生口”——run_in_threadpool。它的适用场景是：你的 handler 大框架是个 async def，大部分逻辑都能 await，偏偏中间夹了一小段必须调同步阻塞函数的代码。这时候，你只需要在那个调用点写一句 await run_in_threadpool(blocking_func, ...)，就能把这次阻塞操作精准地扔到线程池，事件循环在等待它的时候依然可以推进其他请求。

下面这张决策树能帮你在日常开发时快速做出选择：

flowchart TD A[handler 函数体] --> B{存在可 await 异步 I/O?} B -- 是 --> C{有异步驱动?} C -- 是 --> D[async def + await 首选，主推] C -- 否 --> E{其它段也要 await?} E -- 是 --> F[async def + await run_in_threadpool] E -- 否 --> G[def 由框架投递线程池] B -- 否 --> H{纯 CPU?} H -- 是 --> I[def 线程池；大计算走进程池] H -- 否 --> J[def 或 async def 按风格统一]

（图注：决策树一目了然。首选永远是 async def；只有异步驱动实在不可用，或者整个路径都绑死在同步阻塞库上时，才退回到 def。）

为了让你对线程池的工作方式更有体感，这里分别画出 def handler 和 async def + run_in_threadpool 两种模式的执行流：

sequenceDiagram participant C as 客户端 participant S as Uvicorn participant H as Handler participant DB as 数据库 Note over H: def handler（同步阻塞库） C->>S: 请求 S->>H: 投递到线程池 H->>DB: 阻塞读取 DB-->>H: 返回 H-->>S: 响应 S-->>C: 响应 Note over S: 事件循环空闲，可调度其他请求

（图注：def handler 整个在线程池中执行，事件循环全程不受阻。）

sequenceDiagram participant C as 客户端 participant S as Uvicorn participant H as async def Handler participant P as 线程池 participant Sync as 同步阻塞库 Note over H: async def + run_in_threadpool C->>S: 请求 S->>H: 调度协程 H->>P: await run_in_threadpool(_render_pdf, content) P->>Sync: 在工作线程调用 Sync-->>P: 返回字节 P-->>H: 包装为 awaitable H-->>S: 响应 S-->>C: 响应

（图注：async def 内仅阻塞部分被精准投入线程池，事件循环在等待时可调度其他协程。）

代码实现：四段示例，覆盖所有落点

下面用四段代码把前面说的几种情况一次性展示清楚。从纯异步到纯同步，再到混合形态，对照着看会非常直观。数据库相关的 handler 一律走 async def + await，和第 13 章的 AsyncSession 异步栈保持一致；CPU 密集和躲不开的同步库调用，则交给 def 或 run_in_threadpool 兜底。

示例 A：async def 异步并发
这是最理想的状态，两段异步 I/O 用 asyncio.gather 并发调度，总耗时基本等于较长的那一段。

# app/api/books_meta.py（示例 A：async def 异步并发）
import asyncio
from fastapi import APIRouter

router = APIRouter(prefix="/books", tags=["books"])


async def _fetch_meta(book_id: int) -> dict:
    await asyncio.sleep(0.05)
    return {"id": book_id, "title": f"book-{book_id}"}


async def _fetch_stock(book_id: int) -> int:
    await asyncio.sleep(0.01)
    return 1


@router.get("/{book_id}/v2")
async def get_book_async(book_id: int) -> dict:
    """async def + asyncio.gather：并发等待两段异步 I/O，总耗时近似较长的一段。"""
    meta, stock = await asyncio.gather(
        _fetch_meta(book_id),
        _fetch_stock(book_id),
    )
    meta["stock"] = stock
    return meta

示例 B：async def 直连 aiomysql
这里演示的是第 13 章异步栈在路由层的标准形态——通过 Depends(get_async_db) 注入 AsyncSession，然后用 await db.execute(...) 操作数据库。

# app/api/books_db.py（示例 B：async def 真连 aiomysql）
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db import get_async_db
from app.models.book import Book

router = APIRouter(prefix="/books", tags=["books"])
DBDep = Annotated[AsyncSession, Depends(get_async_db)]


@router.get("/{book_id}/db")
async def get_book_db(book_id: int, db: DBDep) -> dict:
    """async def + Depends(get_async_db)：真连 aiomysql 占位读取；完整实现在第 15 章。"""
    result = await db.execute(select(Book).where(Book.id == book_id))
    book = result.scalar_one_or_none()
    return {"id": book_id, "found": book is not None}

示例 C：def 处理纯 CPU 计算
哈希计算是典型的 CPU 密集型操作，没有任何 I/O。这种情况用 def 就好，框架会自动把它放进线程池。

# app/api/hash.py（示例 C：def 纯 CPU 计算）
import hashlib
from fastapi import APIRouter

router = APIRouter(tags=["health"])


def _sha256(payload: bytes) -> str:
    """纯 CPU 计算（哈希），无 I/O，Starlette 自动放入线程池。"""
    return hashlib.sha256(payload).hexdigest()


@router.post("/hash")
def hash_endpoint(payload: bytes) -> dict:
    """def handler：纯 CPU 计算；线程池与阻塞 IO 走同一条路径。"""
    return {"sha256": _sha256(payload)}

示例 D：async def + run_in_threadpool 混合形态
这是最实用的“逃生口”模式。handler 主逻辑是异步的，偏偏有个 PDF 渲染库只提供同步版本，那就用 await run_in_threadpool 把那次调用单独扔进线程池。

# app/api/pdf.py（示例 D：async def + run_in_threadpool）
from starlette.concurrency import run_in_threadpool
from fastapi import APIRouter

router = APIRouter(tags=["pdf"])


def _render_pdf_sync(content: str) -> bytes:
    """模拟同步阻塞 PDF 渲染库。生产中此处会调 wkhtmltopdf / reportlab 等同步 SDK，
    阻塞数十毫秒甚至更久，因此不能直接放进 async def 函数体。"""
    return content.encode("utf-8")


@router.post("/render")
async def render_pdf(content: str) -> dict:
    """async def handler：大部分逻辑能 await，仅 PDF 渲染一段没有异步版本，
    用 await run_in_threadpool 把这一次调用扔到线程池，事件循环在等待时不阻塞。"""
    pdf_bytes = await run_in_threadpool(_render_pdf_sync, content)
    return {"size": len(pdf_bytes)}

避坑指南：这些细节最容易绊倒人

别在 async def 里直接调同步阻塞库。如果不小心在协程里写了个裸的 requests.get(...)，整个事件循环都会被它卡住，所有其他请求的 P99 延迟都会跟着抖动。解决办法就两条：要么把整个 handler 改成 def 让框架帮你投递线程池，要么在调用点包上 await run_in_threadpool(...)。
纯 CPU 计算放在 def 里，但仍受 GIL 限制。Python 解释器同一时刻只有一个线程能执行字节码，线程池在这里只是帮你把任务从主循环里“摘”出来，并不会带来真正的并行加速。计算量一大，还是得上 multiprocessing.Pool 或者 Celery 这类独立任务系统。
run_in_threadpool 是精准手术刀，不是全局开关。它不会改造函数体里其他 await，只把你传进去的那次同步调用扔到线程池。注意线程池的默认容量是 40，超出就会排队。在关键调用点最好显式控制超时时间，别让响应延迟被无限拉长。
asyncio.gather 的异常处理要留个心眼。默认情况下，任意一个协程抛异常都会立刻传播到 await gather(...) 并取消其他协程。如果你需要“不管成败全部跑完再一起看结果”，记得传 return_exceptions=True，然后挨个用 isinstance(result, Exception) 检查。

面试 QA：把原理聊透

Q1 [原理]: FastAPI 同时支持 def 与 async def，两者的判别准则到底是什么？

这个问题其实就一个判断标准：函数体内有没有可以 await 的 I/O，跟执行速度无关。

如果所有 I/O 都有异步版本——比如 httpx.AsyncClient、asyncio.sleep，或者咱们第 13 章用的 AsyncSession 执行 await db.execute(...)——那就首选 async def + await。事件循环能在单请求内并发调度多个 await，这是整个系列异步栈的基石。

如果需要调同步阻塞库（requests、同步数据库驱动，或者没提供异步版的 PDF/加密 SDK），就直接用 def handler。Starlette 会在内部通过 anyio.to_thread.run_sync 把它投进线程池，主事件循环不受影响。

纯 CPU 计算也一样用 def，路径和阻塞 I/O 相同，但 GIL 限制了多线程的并行能力，真要并行还得靠进程池或独立任务系统。

从源码角度看，这个分流发生在 fastapi.routing.run_endpoint_function 里：协程函数直接 await，普通函数走 starlette.concurrency.run_in_threadpool。这个判断只在 handler 的边界做一次，框架既不会扫描你的 async def 函数体，也不会去改写里面的同步调用——这一层的防线，得靠你自己来守。

Q2 [源码]: 异步栈下需要调同步阻塞库时，框架怎么防止阻塞事件循环？

答案可能会让你有点意外：框架不会自动防，它把这个选择权交到了你手里。

async def handler 直接跑在事件循环上，框架没法判断你函数体里某次普通调用是不是阻塞的。所以，如果你直接在协程里写 requests.get(...)，事件循环就会被死死卡住，直到这次网络调用返回，后面所有请求都只能在队列里干等着。FastAPI 不会去扫描或改写 async def 的函数体，这层防护得靠开发者自己。

正确的应对方式就两类。第一类：把整个 handler 写成 def，让 Starlette 自动投递到线程池，适合整个处理链路都绑死在同步库上的情况。不过咱们第 13 章走的异步栈（create_async_engine + AsyncSessionLocal），数据库操作本来就是 await db.execute / db.commit / db.refresh，跟 requests 没关系。第二类更精细：handler 主体是 async def，只是局部需要调一下同步阻塞函数，那就在调用点写 await run_in_threadpool(blocking_func, ...)，把这单次调用精准地丢到线程池，事件循环照常运转。

拿咱们的图书管理 API 举例就很清楚：数据库读写走第 13 章的异步栈，handler 清一色 async def + await db.execute(...)；如果有个第三方 PDF 渲染 SDK 死活只给同步版本，那就在那个端点里保持 async def，但在调用渲染函数的地方写上 await run_in_threadpool(_render_pdf_sync, content)。

小结 & 下篇预告

今天咱们把 async def 和 def 的选型准则彻底讲透了：数据库 handler 首选 async def 跟上异步栈；纯 CPU 或绕不开的同步阻塞库用 def；遇到混合场景，await run_in_threadpool(...) 就是那个精准的逃生口。

下一章《04 路由》会切入真正的工程化实践——APIRouter 怎么拆分、prefix 与 tags 如何优雅地组织，以及咱们图书管理 API 第一组端点的完整注册流程。路由是项目骨架，搭好了后面才能顺风顺水，记得跟上。

FastAPI系列-03-同步与异步