第 16 章 · 异步并发与后台任务
本章目标：搞懂事件循环与
def/async def路径函数的真实执行模型，知道什么时候必须写async，并用 BackgroundTasks 把慢活挪到响应之后。
16.1 并发 ≠ 并行
- 并发（concurrency）：一个人（一个线程/事件循环）在等待 A 的间隙去推进 B——适合 I/O 密集（网络、磁盘、数据库）；
- 并行（parallelism）：多个人同时干活——适合 CPU 密集（大量计算）。
Web 应用的耗时几乎都在 I/O 上，所以 FastAPI 的异步模型收益巨大：一个事件循环线程可以在"等数据库返回"的空档里处理几百个其他请求。
16.2 def vs async def 的真实执行模型
这是 FastAPI 最容易被误解的知识点。官方给出的决策表：
| 你的情况 | 路径函数写法 |
|---|---|
要调用的库需要 await | 必须 async def |
| 调用阻塞式库（多数 DB 驱动、requests） | 用普通 def |
| 不和任何外部通信 | async def |
| 拿不准 | 用普通 def |
背后的机制：
python
import time
from fastapi import FastAPI
app = FastAPI()
# 写法 1：async def 里调用阻塞函数 —— 错误示范！
@app.get("/bad/")
async def bad():
time.sleep(3) # 阻塞了整个事件循环，期间所有请求全部卡死
return {"ok": True}
# 写法 2：普通 def —— FastAPI 自动派发到线程池，不阻塞事件循环
@app.get("/sync-endpoint/")
def sync_endpoint():
time.sleep(3) # 只是占了一个线程池线程
return {"ok": True}
# 写法 3：async def + 真正的异步库 —— 完全不阻塞
@app.get("/good/")
async def good():
await asyncio.sleep(3) # 事件循环在此期间照常调度其他请求
return {"ok": True}规则总结：
async def函数运行在主事件循环里。里面出现任何阻塞调用（time.sleep、同步 DB、requests.get），整个应用的并发能力瞬间归零；- 普通
def路径函数会被 FastAPI 放到外部线程池（Starlette 通过 anyio 的 to_thread）中执行，事件循环不受影响； - 两种写法可以随意混用，FastAPI 会分别正确处理。
同理适用于依赖
依赖函数同样遵循该规则：async def 依赖里别做阻塞调用；同步依赖会进线程池。
16.3 后台任务：响应先回，活儿后干
发邮件、写日志、生成报表……这些操作客户端不必等。BackgroundTasks 让你在响应返回之后继续执行任务：
python
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message: str = ""):
with open("log.txt", mode="a") as f: # 普通阻塞函数也可以
f.write(f"notification for {email}: {message}\n")
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
# 第一个参数是函数本体，后面按位置/关键字传参
background_tasks.add_task(write_notification, email, message="welcome")
return {"message": "Notification sent in the background"}要点：
BackgroundTasks直接作为参数声明类型即可，FastAPI 自动注入；- 任务函数可以是
def或async def，FastAPI 都能正确执行（同步任务在线程池跑）； - 可以连续多次
.add_task()注册多个任务，它们按注册顺序执行； - 执行时机在响应发送之后、且在所有中间件完成之后。
注册多个任务的完整示例：
python
@app.post("/orders/{order_id}/receipt")
async def send_receipt(order_id: int, background_tasks: BackgroundTasks):
# 任务按注册顺序依次执行：先写审计，再发邮件，最后打统计点
background_tasks.add_task(write_audit_log, order_id)
background_tasks.add_task(send_email,
to=f"user-{order_id}@example.com",
template="receipt") # 关键字参数原样透传
background_tasks.add_task(increment_metric, "receipts_sent")
return {"status": "accepted"} # HTTP 202 语义更贴切：已受理，未完成16.4 在依赖里注册后台任务
BackgroundTasks 与依赖注入系统完全打通：路径函数和多层依赖里拿到的是同一个对象，所有注册的任务会合并执行：
python
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", "a") as f:
f.write(message + "\n")
async def query_user(user_id: str, background_tasks: BackgroundTasks):
# 依赖里也能注册：请求级通用逻辑（如审计）放这里最合适
background_tasks.add_task(write_log, f"user {user_id} queried")
@app.get("/users/{user_id}")
async def read_user(
user_id: str,
background_tasks: BackgroundTasks,
_: None = Depends(query_user),
):
background_tasks.add_task(write_log, f"response sent for {user_id}")
return {"user_id": user_id}16.5 边界：BackgroundTasks vs Celery
官方文档明确给出了选择标准：
- BackgroundTasks 适用：轻量任务；需要访问同一进程内的对象/内存/应用状态；不想引入额外基础设施（消息队列）；
- Celery/RQ/Dramatiq 等任务队列适用：重计算；需要跨进程、跨机器扩展；需要持久化重试、定时调度、任务结果查询——代价是要部署 RabbitMQ/Redis 等 broker。
一句话：BackgroundTasks 是"进程内延迟执行"，进程重启它就没了；任务队列才是分布式作业系统。两者不冲突，很多生产系统两者并存。
16.6 本章小结
- 并发是"等待时切换"，并行是"同时多干"；Web 场景主要是 I/O 并发；
async def跑在事件循环上，里面严禁阻塞调用；普通def自动进线程池；拿不准就用def；BackgroundTasks.add_task(func, *args, **kwargs)在响应后顺序执行注册的任务，支持在依赖中注入并合并；- 重计算/跨机扩展选 Celery 类队列，轻量同进程任务用 BackgroundTasks。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 在 async def 路径函数中直接调用 time.sleep(3)，后果是什么？
2. 普通 def 定义的路径函数在 FastAPI 中如何执行？
3. 关于 BackgroundTasks，下列说法错误的是？
4. 以下哪种需求最适合用 Celery 而不是 BackgroundTasks？
🛠️ 动手实践
- 写三个端点分别复现 16.2 的错误/线程池/真异步三种写法，用浏览器开两个标签同时请求，观察阻塞差异并记录响应时间。
- 实现"注册接口"：POST /register 校验通过后立刻返回 202，用两个后台任务分别写审计日志和模拟发邮件（sleep 3 秒打印），验证响应不被拖慢。
- 给第 13 章的用户注册端点加后台密码哈希强度测试（用 pwdlib 对假哈希 verify 一次计时），思考这个例子说明哈希为什么必须放后台还是可以同步做，写出结论。
学会让数据"主动说话"了吗？请进入下一章：WebSockets 实时通信。