Chapter 07 · 异步与后台任务：不等人的艺术
📚 学完本章，你将能够：
- 用生活化的比喻理解 async/await 和事件循环
- 判断一个操作是否应该写成异步的
- 用
BackgroundTasks处理发邮件等轻量后台工作 - 用
asyncio.gather并发执行多个 I/O 操作 - 识别 CPU 密集型任务，并知道为什么它不适合放在 async 里
💡 场景引入
用户在你的网站注册账号。注册逻辑包括：写数据库 + 发欢迎邮件。如果发邮件需要 3 秒钟，用户就得盯着浏览器转圈 3 秒才能看到"注册成功"。
这显然不合理。发邮件不应该让用户等。本章教你如何让"发邮件"这件事在后台悄悄进行，同时用异步技术让你的服务器能同时服务更多用户。
7.1 同步 vs 异步：一个咖啡店的比喻
想象你是一个咖啡师（服务器），面前有一排顾客（请求）：
同步模式：一个顾客做完才做下一个
def make_coffee(order):
grind_beans() # 2秒
brew() # 3秒
serve() # 1秒
# 顾客A（6秒）→ 顾客B（6秒）→ 顾客C（6秒）
# 总耗时：18秒你在等咖啡豆磨好的时候，干站着不服务其他人。这就是同步阻塞。
异步模式：磨豆的时候去接待下一个人
async def make_coffee(order):
await grind_beans() # "我去磨豆了，磨好叫我"
await brew() # "我去煮了，煮好叫我"
serve()
# 三个顾客的请求交错执行
# 总耗时：约 6 秒（不是 18 秒！）await 的意思是"这件事需要等待，你去忙别的吧"。事件循环会在磨豆完成时自动切回来。
核心区别：同步代码阻塞线程，异步代码释放线程去干别的。但只有I/O 操作（网络请求、数据库查询、文件读写）才适合异步。如果是纯计算（比如算圆周率后100位），await 不会释放线程，因为 CPU 根本停不下来。
7.2 FastAPI 中的 BackgroundTasks
对于"发邮件"这种不需要立刻完成、也不要求结果的操作，FastAPI 提供了最简单的后台任务机制：
后台发邮件
from fastapi import BackgroundTasks
async def send_welcome_email(email: str, username: str):
# 模拟发邮件耗时 2 秒
await asyncio.sleep(2)
print(f"邮件已发送至 {email}")
@app.post("/register")
async def register(
data: UserCreate,
background_tasks: BackgroundTasks,
db: Annotated[AsyncSession, Depends(get_db)]
):
user = await create_user(db, data)
# 把发邮件任务丢到后台，不阻塞响应
background_tasks.add_task(send_welcome_email, user.email, user.username)
# 用户立刻收到响应，不需要等邮件发完
return {"message": "注册成功", "user_id": user.id}
BackgroundTasks 的限制
它运行在 Uvicorn 的同一个进程里。如果任务很重（比如处理大文件），或者任务失败了，没有重试机制。对于重要的后台工作，需要用 Celery（第 11 章介绍）。
7.3 并发查询：asyncio.gather
假设一个页面需要同时展示：用户信息 + 订单列表 + 优惠券。如果串行查询，总耗时是三部分之和。用 gather 可以并发发请求：
并发查询
import asyncio
@app.get("/dashboard")
async def dashboard(user_id: int, db: Annotated[AsyncSession, Depends(get_db)]):
# 三个查询同时发出去，总耗时 ≈ 最慢的那个
user, orders, coupons = await asyncio.gather(
get_user(db, user_id),
get_orders(db, user_id),
get_coupons(db, user_id),
)
return {"user": user, "orders": orders, "coupons": coupons}串行 vs 并发
串行（慢）
用户查询 100ms 订单查询 150ms → 总耗时 350ms 优惠券查询 100ms
并发（快）
用户查询 ──┐ 订单查询 ──┼──→ 总耗时 150ms 优惠券查询 ──┘
7.4 CPU 密集型任务的陷阱
asyncio 不能加速计算。如果你在协程里做大量数学运算，整个事件循环会被卡住，其他请求都得等着。
❌ 错误：在 async 里做 CPU 计算
@app.post("/analyze")
async def analyze(data: list[float]):
# 这个循环会阻塞事件循环！
result = sum(x ** 2 for x in data) # 假设数据很大
return {"result": result}解决方案：用 ProcessPoolExecutor 把计算丢到单独的进程里，释放事件循环：
✅ 正确：进程池卸载
from concurrent.futures import ProcessPoolExecutor
import asyncio
pool = ProcessPoolExecutor(max_workers=4)
def heavy_compute(data: list[float]) -> float:
return sum(x ** 2 for x in data)
@app.post("/analyze")
async def analyze(data: list[float]):
# 把 CPU 任务丢到进程池，async 事件循环继续处理其他请求
result = await asyncio.get_event_loop().run_in_executor(pool, heavy_compute, data)
return {"result": result}🎯 知识检测
问题 1
以下哪种操作最适合用 async/await？
解析：async/await 的优势在于 I/O 等待时释放线程。数据库查询、网络请求、文件读写都是 I/O 操作。而纯计算（数学运算、视频压缩、图像渲染）是 CPU 密集型，async 不会加速它们，反而需要用进程池卸载。
问题 2
BackgroundTasks 适合做什么？
解析：BackgroundTasks 运行在 Uvicorn 进程内，没有重试机制，进程崩溃时任务会丢失。它只适合轻量、非关键的后台工作。关键任务用 Celery，大计算用专门的 worker。
问题 3
asyncio.gather 的作用是什么？
解析：gather 让你同时发多个异步请求（如查用户、查订单、查优惠券），总耗时等于最慢的那个任务，而不是所有任务之和。这是提升 API 响应速度的利器。
本章小结
✓
async 只加速 I/O 等待
数据库查询、网络请求、文件读写适合 async。纯计算不适合。
✓
BackgroundTasks 做轻量后台工作
发邮件、记日志，不阻塞用户响应。但不保证一定成功。
✓
gather 并发多个 I/O 操作
多个查询同时发，总时间由最慢的决定，大幅提升响应速度。
✓
CPU 任务用进程池卸载
别把重计算放在 async 函数里，会卡住整个服务器。