并发与阻塞代码

上一章我们说了，FastAPI 默认推荐 async def，靠 await 在等待时让出控制权。但事情有个前提：只有”能被 await 的慢操作”才能这么让位。如果你的代码里有一步是卡住不让位的，整个事件循环就会被它拖住。这一章就来解决这类”阻塞代码”。

39-1 什么是阻塞代码

阻塞代码指的是：执行到这一步，程序必须老老实实等它跑完，期间什么别的请求都顾不上。典型的两种：

第一种是阻塞 I/O。比如你用了一个不支持 await 的库去读文件、发请求、查数据库。这个库内部是同步的，调用它的时候线程就被占住了。

第二种是 CPU 密集计算。比如你要处理一张大图、做大量数学运算、跑机器学习推理。这类活儿占的是 CPU，不是”等”，所以即使写了 await 也没用——因为根本没有 await 可让位，CPU 一直在忙。

39-2 阻塞 I/O：直接写成 def 就行

最省心的办法：如果你的路径操作里要调用不支持异步的库，那就把函数写成普通的 def。FastAPI 看到 def，会自动把它放进一个外部线程池里执行。线程池里的线程被卡住没关系，主事件循环照常处理别的请求。

import time
from fastapi import FastAPI

app = FastAPI()


@app.get("/sync-read")
def sync_read():
    # 假设这是一个不支持 await 的同步库调用
    time.sleep(1)
    return {"message": "done with blocking call"}

这样写，FastAPI 在背后用线程池兜住了阻塞，你的接口依然能并发响应其他用户。你不需要自己管线程，def 就够了。

39-3 CPU 密集任务：也建议用 def

CPU 密集任务有个坑：如果你写在 async def 里，但里面没有 await，那么这段计算会一直霸占事件循环，其他所有请求都得排队。所以在 async def 里硬算是个错误示范（上一章说的“轻量计算写 async def”指的是毫秒级、算得很快的那种；这里说的是耗时较长的重计算）。

正确做法是写成 def，让 FastAPI 丢进线程池，至少不会卡死事件循环。需要注意的是，纯 Python 计算受 GIL 限制，线程池并不能真正利用多核；想真正跑满多核，得用多进程（ProcessPoolExecutor）或另起服务：

from fastapi import FastAPI

app = FastAPI()


def heavy_compute(n: int) -> int:
    # 假装是个很费 CPU 的计算
    total = 0
    for i in range(n):
        total += i
    return total


@app.get("/compute")
def compute():
    result = heavy_compute(10_000_000)
    return {"result": result}

39-4 run_in_threadpool 登场

有时你会遇到这种情况：路径操作整体是异步的，你已经用了 async def，但中间偏偏要调一个同步阻塞函数。直接调会卡住事件循环，写 def 又得改整个函数签名。

FastAPI 提供了 run_in_threadpool，让你在 async def 内部，把阻塞函数丢到线程池去跑，自己这边 await 着等结果。一举两得：

from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool

app = FastAPI()


def blocking_task(name: str) -> str:
    # 一个不支持 await 的同步阻塞函数
    import time
    time.sleep(1)
    return f"finished {name}"


@app.get("/task")
async def run_task():
    # 把阻塞函数交给线程池，await 等它完成
    result = await run_in_threadpool(blocking_task, "job-1")
    return {"result": result}

run_in_threadpool 的第一个参数是那个函数，后面跟的是它要的位置参数和关键字参数。调用时前面加 await，FastAPI 会在后台线程里执行它，事件循环趁机去处理别的请求。

39-5 一个更完整的例子

下面这个例子混合了异步和阻塞：先 await 一个异步 HTTP 调用，再用 run_in_threadpool 跑一个同步的文件读取。两种等待都不阻塞主循环。

import httpx
from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool

app = FastAPI()


def read_config_file() -> str:
    with open("config.txt", "r", encoding="utf-8") as f:
        return f.read()


@app.get("/mixed")
async def mixed():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://example.com/status")

    config = await run_in_threadpool(read_config_file)
    return {"status": resp.status_code, "config": config}

这里异步部分用 await，同步部分用 run_in_threadpool，两条路都走通了。代码清晰，也不卡服务器。

39-6 async 与线程池怎么取舍

把几条规则串起来，给你一张速查表：

• 调支持 await 的库（异步数据库、异步 HTTP）：用 async def + await。这是最理想的。
• 调不支持 await 的库（多数传统库）：用 def，FastAPI 自动丢线程池。
• 在 async def 里非要调同步库：用 run_in_threadpool 包一层。
• 纯 CPU 计算：用 def（线程池），别在 async def 里裸算。
• 实在拿不准：用 def，永远安全。

你可能会问：那 async def 比 def 快多少？其实差别很小。写 async def 的真正收益，是当你整条调用链都是异步库时，事件循环能丝滑地并发。如果链路里混着阻塞库，那优势就被吃掉了。所以取舍的核心不是”async 更快”，而是”别让阻塞卡住事件循环”。

39-7 小结

这一章我们搞定了阻塞代码：

• 阻塞 I/O 和 CPU 密集都会卡住事件循环，需要特殊处理。
• 最简单的办法是把路径操作写成 def，FastAPI 自动丢进线程池。
• 在 async def 内部想调同步函数，用 run_in_threadpool 包一层再 await。
• 选法的核心是”别阻塞”，而不是盲目追求 async。

下一章我们讲后台任务（BackgroundTasks），它专门用来处理”响应已经返回，但还有些收尾活要干”的场景。