FastAPI-05补.异步原理实现与踩坑手册

发表于 | 分类于 ，
工具书：asyncio 事件循环与协程原理、FastAPI/Starlette 对 sync/async 路由的调度实现、真假异步、线程池/取消/依赖与中间件注意点，以及可检索的踩坑清单与决策表。

本系列：00 导读 · 01补 网络基础 · 01 心智模型 · 02 路由与数据模型 · 03 依赖注入与分层 · 04 中间件异常日志 · 05 异步后台与流式 · 05补 异步原理与踩坑（本文） · 06 鉴权与安全 · 07 测试与项目骨架 · 08 实战 HTTP↔MCP

行文：T1 + T3（工具书体）　｜　本篇方法：第一性原理 + 费曼　｜　辅助：双重编码、交错排错

定位：05 讲「怎么用」与组块练习；本文讲「为什么这样调度、实现落在哪一层、哪些写法必炸」。查表式阅读，不必一次读完。

社区方案优先：异步 I/O 以标准库 asyncio 为合同；HTTP 客户端优先 httpx（AsyncClient）；DB 优先驱动自带的 async API（如 SQLAlchemy 2.0 async、databases、asyncpg）。独立长任务优先 ARQ / Celery / RQ 等队列，而不是自造「全局 asyncio 任务池」。潜在风险见各节「风险」小节。

0. 速查：一张决策表

你要做的事 推荐写法 忌讳
等网络 / 等 DB（真异步库） async def + await 在 async def 里调同步 requests / 同步 ORM
遗留同步库、短阻塞 IO def 路由（框架丢线程池）或 asyncio.to_thread 假装已异步
CPU 重算、大 JSON 序列化、图像处理 to_thread / 进程池 / 独立 worker 在事件循环上直接算
响应后轻量收尾（写审计日志） BackgroundTasks 支付落账、必须成功的副作用
长任务、可重试、跨进程 Celery / ARQ 等 只靠进程内 create_task
流式 / SSE / LLM token StreamingResponse + async 生成器 同步 time.sleep 堵循环

段末注释：asyncio 是 Python 标准库异步 I/O 与事件循环框架；事件循环（event loop）在单线程内调度协程，等 I/O 时切换执行权。后文沿用缩写与简称。

1. 问题：没有「异步心智」会坏在哪

现象 常见根因
QPS 一上来全站超时，CPU 却不高 事件循环被同步阻塞占满
写了 async def 却不比 Flask 快 库仍是同步的（假异步）
偶发「任务没跑完响应已返回」 误用 create_task 且未管生命周期/异常
依赖里拿不到同一请求的上下文 跨线程后 ContextVar 丢失
客户端取消请求后服务端仍狂写库 未处理 CancellationError / 未感知 disconnect
多 worker 下「全局单例连接」错乱 把进程级资源当成全集群唯一

本篇把原理、框架实现、注意事项拆开，便于对照排障。

2. 第一性原理：异步到底在优化什么

2.1 并发 ≠ 并行

  • 并发（concurrency）：一段时间内推进多件事（可交错）。
  • 并行（parallelism）：同一时刻真的在多核上同时算。

asyncio 默认模型是：单线程并发——适合「大量等待」（网络、磁盘、远端 API）。CPU 密集需要并行时，要额外用线程/进程/多 worker，而不是多写几个 async def。

可用等待占比粗估收益：若单请求耗时中「可等待 I/O」占比为 (p)，协作式调度在理想情况下能让同线程吞吐接近同步串行的 (1/(1-p)) 量级（直觉公式，非严格排队论）。(p) 很小（纯 CPU）时，异步几乎无增益。

2.2 协程、Task、事件循环

概念 是什么 谁创建
协程对象 async def 调用后得到的 awaitable 你调用 foo()（未 await 则未跑）
Task 已调度进循环的协程包装 asyncio.create_task / 框架内部
事件循环 取就绪回调、推进 Task、处理 I/O uvicorn 在 worker 内持有

关键句：只有 await（或等价挂起点）才会把控制权交还循环。 协程里若全程同步阻塞，对循环而言等价于「一个永不让出的占用者」。

2.3 费曼白话

把事件循环想成单车道圆盘传送带：每个协程是工人。工人说「我要等快递」（await），就站到旁边，传送带可以转给别人。若有人在传送带上抡大锤砸螺丝（同步阻塞），整条带停转，后面所有请求一起堵。

图 A　事件循环：协程让出 vs 同步阻塞卡住

3. 实现：从 asyncio 到 FastAPI 的调度链

3.1 ASGI 服务器持有循环

uvicorn main:app 大致做：

  1. 建事件循环（或使用 uvloop 等实现）。
  2. 跑 lifespan（启动/关闭）。
  3. 每个 HTTP 连接：构造 scope，调用 ASGI 应用 await app(scope, receive, send)。

详见 01 心智模型。异步不是 FastAPI 发明的；FastAPI/Starlette 只是在合同上组织路由与依赖。

3.2 Starlette：async def 与 def 分流（必记）

Starlette 对端点函数大致策略：

端点定义 调度 含义
async def 在事件循环上直接 await 必须全程可协作；阻塞即堵全站
def（同步） 丢进线程池执行 阻塞 IO 不堵循环，但占线程、有切换成本

依赖函数（Depends）同样遵循「async 走循环、sync 走线程池」的同类规则（实现细节以当前 Starlette 版本为准，心智模型不变）。

图 B　async 路由进循环，def 路由进线程池；假异步堵循环

工具书结论：

  • IO 密集 + 有异步库 → async def。
  • 只有同步库、调用短 → def 往往更安全省事。
  • 最差组合：async def + 同步阻塞库。

3.3 最小对照代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import time

import httpx
import requests
from fastapi import FastAPI

app = FastAPI()


@app.get("/ok-async")
async def ok_async():
    """真异步：await 期间循环可服务其他请求。"""
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.get("https://httpbin.org/delay/1")
    return {"status": r.status_code}


@app.get("/bad-fake-async")
async def bad_fake_async():
    """假异步：同步 requests 卡住整个事件循环。"""
    r = requests.get("https://httpbin.org/delay/1", timeout=10)
    return {"status": r.status_code}


@app.get("/ok-sync-threadpool")
def ok_sync_threadpool():
    """同步路由：阻塞发生在线程池，不直接堵循环。"""
    r = requests.get("https://httpbin.org/delay/1", timeout=10)
    return {"status": r.status_code}


@app.get("/cpu-wrong")
async def cpu_wrong():
    """CPU 重活直接跑在循环上：其他请求饥饿。"""
    n = sum(i * i for i in range(5_000_000))
    return {"n": n}

压测时对比 /ok-async 与 /bad-fake-async：后者在并发下会拖垮同 worker 的其他接口延迟。

3.4 把阻塞挪出循环的三种实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import asyncio
from concurrent.futures import ProcessPoolExecutor

from fastapi import FastAPI

app = FastAPI()


def sync_io_call(url: str) -> int:
    import requests
    return requests.get(url, timeout=10).status_code


def cpu_bound(n: int) -> int:
    return sum(i * i for i in range(n))


@app.get("/to-thread")
async def via_to_thread():
    """Python 3.9+：把同步函数丢默认线程池。"""
    code = await asyncio.to_thread(sync_io_call, "https://httpbin.org/get")
    return {"status": code}


@app.get("/executor-cpu")
async def via_process_pool():
    """CPU 密集优先进程池，避开 GIL 争用（注意序列化成本）。"""
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        result = await loop.run_in_executor(pool, cpu_bound, 5_000_000)
    return {"n": result}
API 适用 风险
asyncio.to_thread(fn, ...) 同步 IO、轻量 CPU 默认线程池有上限；线程过多反而抖
loop.run_in_executor(None, ...) 同 to_thread 的老写法 勿对已关闭的 loop 提交
ProcessPoolExecutor CPU 密集 参数/返回值须可 pickle；进程启动贵

段末注释：GIL（全局解释器锁，Global Interpreter Lock）使多线程对纯 Python CPU 计算加速有限；CPU 密集常需多进程或多服务实例。

4. 协作原语速查（在 FastAPI 里怎么用）

4.1 await / 并发扇出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import asyncio

import httpx
from fastapi import FastAPI

app = FastAPI()


@app.get("/fan-out")
async def fan_out():
    async with httpx.AsyncClient(timeout=10.0) as client:
        # 并发发起，总耗时约等于最慢的那个，而非相加
        r1, r2 = await asyncio.gather(
            client.get("https://httpbin.org/delay/1"),
            client.get("https://httpbin.org/delay/1"),
        )
    return {"a": r1.status_code, "b": r2.status_code}
  • gather：默认一失败全取消（可 return_exceptions=True）。
  • TaskGroup（3.11+）：结构化并发，作用域退出时收束子任务，推荐新代码优先。

4.2 create_task：火后不管的雷区

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import asyncio

from fastapi import FastAPI

app = FastAPI()


async def side_effect():
    await asyncio.sleep(2)
    # 若此处抛错且无人 await，可能变成「未检索的异常」日志
    ...


@app.post("/fire-and-forget-bad")
async def bad():
    asyncio.create_task(side_effect())  # 请求返回后任务仍跑，但难观测、易丢
    return {"ok": True}

规范替代：

需求 做法
响应后轻量活 BackgroundTasks（见 05）
必须可靠 外置队列 + worker
请求内并行 gather / TaskGroup，并在返回前 await 完

4.3 超时与取消

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import asyncio

import httpx
from fastapi import FastAPI, HTTPException

app = FastAPI()


@app.get("/with-timeout")
async def with_timeout():
    try:
        async with httpx.AsyncClient() as client:
            async with asyncio.timeout(2.0):  # 3.11+
                await client.get("https://httpbin.org/delay/5")
    except TimeoutError:
        raise HTTPException(status_code=504, detail="upstream timeout")
    return {"ok": True}

客户端断开时，ASGI 服务器可能取消当前 Task。业务里若吞掉所有异常（裸 except Exception）可能吞掉 CancellationError 的基类路径（3.8+ CancelledError 继承 BaseException，但仍禁止「乱捕后不重抛」的习惯）。长事务应设计为可中断或幂等。

4.4 ContextVar 与线程边界

  • 请求级上下文（追踪 ID、当前用户）常用 contextvars。
  • def 路由在线程池跑时，框架通常会复制上下文；自己 to_thread / 自建线程时，不要假设 ContextVar「自动过去」——需要显式传参或 contextvars.copy_context().run(...)。

5. FastAPI 特有集成点

5.1 依赖注入：async / sync 混用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from fastapi import Depends, FastAPI

app = FastAPI()


async def get_conn():
    # 异步依赖：在事件循环上执行
    return {"db": "async"}


def get_settings():
    # 同步依赖：可能进线程池；保持轻量（读配置 OK，别塞重 IO）
    return {"env": "prod"}


@app.get("/item")
async def item(
    conn=Depends(get_conn),
    settings=Depends(get_settings),
):
    return {"conn": conn, "settings": settings}

注意：同步依赖若做重阻塞，会占线程池配额，表现为「异步路由也排队」。依赖保持薄，重 IO 放进显式的 repository 层并选对 async/sync。

5.2 中间件

  • 纯 ASGI 中间件（async def __call__(self, scope, receive, send)）与循环同生共死，禁止内部同步阻塞。
  • @app.middleware("http") 的 async 函数同样跑在循环上。
  • 认证解析、日志、计时放中间件可以；调同步 HTTP、同步 DB 不行（或先 to_thread）。

5.3 lifespan 与连接池

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.client = httpx.AsyncClient(timeout=10.0)
    yield
    await app.state.client.aclose()


app = FastAPI(lifespan=lifespan)
  • 每 worker 一份池/客户端；多进程不是「全局一份」。
  • 启动失败应让进程退出，避免半开服务。
  • 关闭时 await 释放，避免 fd 泄漏。

5.4 WebSocket / 流式

  • WebSocket 端点天然长生命周期，receive/send 循环内同样禁止同步阻塞。
  • StreamingResponse 的 async 生成器里用 await asyncio.sleep，不用 time.sleep。细节与 SSE 见 05 §5。

6. 真异步生态：库怎么选

场景 优先社区方案 忌用（在 async 路由内） 风险备注
HTTP 出站 httpx.AsyncClient、aiohttp requests 每请求新建 Client 有开销；优先 lifespan 复用
PostgreSQL asyncpg / SQLAlchemy async 同步 psycopg2 直调 连接池大小按 worker×池 估算
Redis redis.asyncio 同步 redis-py 阻塞调用 注意 decode 与超时
文件 aiofiles 或 to_thread 大文件同步 open().read() 超大文件考虑流式与磁盘压
ORM SQLAlchemy 2.0 AsyncSession 在 async 里调同步 Session 「假异步 ORM」极常见
任务队列 ARQ（asyncio 友好）、Celery 无限 create_task 队列有运维成本，但语义清晰

潜在风险（采用社区方案时仍要注意）：

  1. 版本矩阵：SQLAlchemy async、驱动、greenlet 组合需对齐。
  2. 连接风暴：workers × pool_size 打满数据库 max_connections。
  3. httpx 连接池：未关闭 Client 导致 fd 泄漏；测试里尤其常见。

7. 踩坑手册（错误 → 现象 → 根因 → 修法）

图 C　假异步 / CPU 堵循环 / BackgroundTasks 语义边界

# 错误写法 现象 根因 修法
1 async def + requests / 同步 ORM 延迟尖刺、级联超时 阻塞事件循环 换异步库，或改 def / to_thread
2 async def + time.sleep 同上 同 #1 await asyncio.sleep
3 async def + 重 CPU 其他接口饿死 循环无法调度 进程池 / 独立服务
4 忘记 await 协程 警告、逻辑没执行 只创建了协程对象 必须 await 或显式建 Task 并托管
5 create_task 火后不管 偶发丢活、异常难查 无生命周期与监督 BackgroundTasks 或队列
6 BackgroundTasks 做支付 重启丢单 进程内内存任务 持久化队列 + 幂等
7 同步中间件调远端 全站卡顿 中间件在循环路径上 异步客户端或拿掉
8 线程里碰非线程安全对象 偶现坏数据 对象绑定循环/线程 传原始数据；每线程自建客户端
9 裸 except: 吞取消 关停变慢、任务僵死 取消被吞 捕获后 raise；区分业务异常
10 多 worker 共用「全局 dict 缓存」当一致存储 读到旧数据 多进程内存不共享 Redis 等外置缓存
11 async 路由里同步读超大文件 卡顿 磁盘阻塞 aiofiles / 线程 / 流式
12 def 里写 await 语法错误 sync 函数不能 await 改 async def 或去掉 await
13 测试用同步 Client 测异步副作用时序 偶发失败 事件循环与线程交织 用 AsyncClient（httpx）与 pytest-asyncio
14 lifespan 外创建 AsyncClient 运行中改全局 生命周期不清 放入 app.state + lifespan
15 无限积压 gather 上千路 内存与 fd 爆 无并发度限制 Semaphore 限流

7.1 「假异步」判定清单（上线前勾选）

  • 所有 await 的对象来自文档标明 async 的 API
  • 没有 requests / 同步 http.client / 同步 ORM Session
  • 没有 time.sleep、同步 subprocess 长等待
  • CPU 段已隔离或可接受极短
  • 出站 Client / DB 池在 lifespan 管理
  • 压测过：单 worker 下混合慢接口时，快接口延迟是否被拖死

8. 性能与容量：实现层面的数字直觉

旋钮 作用 误区
uvicorn --workers 多进程，各有独立循环 worker 不是「异步开关」；内存与连接数倍增
线程池大小 限制 sync 路由并发 过大则上下文切换与连接打爆下游
DB pool_size 每进程连接上限 总连接 ≈ workers × pool
httpx limits 出站并发与 keepalive 默认未必适合高扇出
超时 防止慢请求占坑 无超时 = 隐性死锁

粗估：异步的收益来自等待重叠；瓶颈在 CPU 或单下游锁时，先扩容下游或拆服务，而不是继续「加 async」。

9. 与 05 篇的分工（避免重复阅读）

主题 看哪篇
BackgroundTasks vs Celery 选型表 05 §4
StreamingResponse / SSE / LLM 流 05 §5
ASGI scope/receive/send、lifespan 时钟 01
原理、分流实现、踩坑全表 本文

10. 合书自测

  1. 用两句话区分：并发与并行；说明 asyncio 默认优化哪一种。
  2. Starlette 对 async def 与 def 端点的调度差异是什么？哪一种写法最危险？
  3. 列举三个「假异步」实例，并各给一种修法。
  4. 为什么说 BackgroundTasks 不适合「必须成功」的账务？社区替代是什么？
  5. 4 个 worker、每进程池 10 连接，数据库至少要准备多少连接额度（忽略管理连接）？

11. 闪卡候选

正面 背面
await 的本质？ 让出事件循环，等待可等待对象完成
假异步定义？ async 函数里仍跑同步阻塞调用
sync 路由跑哪？ Starlette 线程池
CPU 密集首选？ 进程池 / 多进程 worker / 独立服务
create_task 风险？ 生命周期与异常无人接管
取消被吞的后果？ 关停困难、资源不释放
httpx 客户端放哪？ lifespan + app.state，避免每请求泄漏

小结

  • 原理：asyncio 用单线程事件循环做等待重叠；await 才是让出点。
  • 实现：uvicorn 持循环；Starlette 让 async def 跑循环、def 跑线程池；FastAPI 在此之上叠依赖与路由。
  • 注意：假异步、CPU 堵环、火后不管的 Task、多 worker 资源倍增、取消与 ContextVar，是最高频事故源。
  • 用法组块回 05；下一主线 06 鉴权与安全。
-------------本文结束感谢您的阅读-------------