KEEL · 龙骨 · A CURRICULUM FOR THE AI ERA

02 · 异步正确性：阻塞点识别与协程/线程隔离 — keel 龙骨

这一章回答：为什么同一个 FastAPI 服务，加了一个 time.sleep 或一把 threading.Lock，整个接口的延迟就崩了。

这一章回答：为什么同一个 FastAPI 服务，加了一个 time.sleep 或一把 threading.Lock，整个接口的延迟就崩了。

异步框架的性能来自一件事：事件循环上的每个协程都不能长时间占用 CPU 或阻塞等待。只要有一处阻塞，同一循环上的所有请求都会被拖住——这是异步服务最隐蔽也最常见的性能事故。

一、阻塞的两种形态

形态	表现	例子
CPU 阻塞	大量计算占着 CPU 不释放	大 JSON 序列化、压缩、加密
IO 阻塞（同步调用）	等待期间没有 `await`，事件循环无法切换	`time.sleep`、`requests`、同步驱动如 `pymysql`、`redis-py` 同步客户端

关键区别：await 是主动让出控制权，同步调用是占着不放。

二、阻塞点清单（自查用）

在 FastAPI / asyncio 项目里，以下都是必须处理的阻塞点：

类别	危险写法	正确写法
休眠	`time.sleep(3)`	`await asyncio.sleep(3)`
HTTP 调用	`requests.get(...)`	`httpx.AsyncClient` 或 `aiohttp`
数据库	`pymysql`、`SQLAlchemy` 同步会话	`asyncmy` / `asyncpg` + `AsyncSession`
Redis	`redis.Redis`（同步）	`redis.asyncio.Redis`
文件 IO	`open().read()`	`run_in_executor` 或 `aiofiles`
CPU 密集	直接算	`run_in_executor`（进程池更好）
第三方 SDK	只提供同步接口	线程池包装（见下）

三、无法避免的阻塞：隔离而不是消灭

现实里总有只提供同步接口的依赖（老 SDK、某些数据库驱动）。做法是把它关进线程池，让它在另一个线程里阻塞，不占用事件循环：

from concurrent.futures import ThreadPoolExecutor
import asyncio

executor = ThreadPoolExecutor(max_workers=10)  # 按阻塞型调用的并发上限配置

async def call_blocking_sdk(payload):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, blocking_sdk_call, payload)

要点：

max_workers 是这个线程池的并发上限——如果设太小会成为瓶颈，设太大又会对下游造成压力，需要按依赖的承受能力定；
这类调用应该可以被限流与超时包裹，见下一章；
CPU 密集任务用 ProcessPoolExecutor 更合适，线程池在 Python 里受 GIL 限制拿不到并行。

四、锁：`threading.Lock` 在协程里是禁品

这是异步项目里最容易埋错的坑：

场景	该用什么	原因
纯多线程/多进程临界区	`threading.Lock()`（必须 `with`，防死锁与泄漏）	阻塞的是线程，不影响事件循环
协程之间的临界区	`asyncio.Lock()`（`async with`）	它会将控制权交还事件循环
协程里用 `threading.Lock`	❌ 绝对禁止	阻塞的是整个事件循环，同进程所有请求一起卡死

判断方法：看持锁临界区里有没有 await。 有 await 就必须是可等待的锁。

配套三条实践：

缩短持锁时间：锁内不做事先能做完的计算、不做网络调用；
统一加锁顺序：多把锁时按固定顺序获取，避免死锁；
可重入用 asyncio 场景的 RLock，但先确认是不是设计上就不需要重入。

五、一个真实的体感实验

@app.get("/sleep1")     # 危险：阻塞事件循环
async def sleep1():
    time.sleep(1)
    return {"ok": True}

@app.get("/normal")
async def normal():
    return {"ok": True}

用压测工具并发打 /sleep1，同时请求 /normal：

单 worker 下：/normal 的延迟会被拉到约 1 秒，因为它也在同一个事件循环里排队；
改成 await asyncio.sleep(1) 后：/normal 立刻恢复到毫秒级；
改成 run_in_executor 包装后：/normal 同样不受影响，但注意线程池打满时仍会成为新的瓶颈。

这个实验值得亲手做一遍——"一个错误写法拖垮全部请求"这件事，看代码不够直观，看曲线才印象深刻。

动手：可观察结果

产出	判断标准
一份阻塞点自查清单	扫完项目里全部 `sleep`、`requests`、同步客户端、同步 ORM 会话，逐项标记处理状态
一次体感实验	压测 `/sleep1` 的同时请求 `/normal`，记录 P99 延迟；修复后再测，附两次数据
锁选型检查	grep 出全部 Lock 用法，确认协程路径上没有 `threading.Lock`

完成标志：在并发压测下，任意一个慢/阻塞接口的延迟不会影响其他接口的 P99。

故障注入

注入方式	观察	修复方式
在 `async def` 里写 `time.sleep(1)`	同进程其他接口延迟同步升高	改 `await asyncio.sleep`
在协程里误用 `threading.Lock`	请求堆积直至超时	改 `asyncio.Lock`
线程池 `max_workers=1` 后打并发	该依赖成为串行瓶颈	按依赖吞吐调参 + 超时保护
保留一个同步 ORM 会话不放	高并发下连接池耗尽	改异步驱动或线程池隔离

自测题

为什么 time.sleep 在单 worker 的 FastAPI 里会拖慢所有请求，而多 worker 只影响其中一个？
判断该用 threading.Lock 还是 asyncio.Lock 的具体标准是什么？
run_in_executor 解决的是什么问题？它没有解决什么问题（提示：线程池自身）？
CPU 密集任务为什么应该考虑进程池而不是线程池？
列举你项目里三个潜在的阻塞点，并说明你会怎么验证它们确实是阻塞的。

进入 keel 阅读