FastAPI-05.异步后台任务与流式响应

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

行文：T1 + T3　｜　本篇方法：第一性原理 + 费曼　｜　辅助：双重编码、刻意练习

1. 问题：`async def` 写了，为什么还是卡

根因常是：事件循环被阻塞。ASGI 服务器（uvicorn）在单 worker 内用 asyncio 事件循环调度协程；任何同步阻塞（time.sleep、同步 requests.get、重 CPU）会占住循环，其他请求一起等。

写法	实际行为
`async def` + `await httpx.AsyncClient()`	真异步 IO
`async def` + `requests.get()`	假异步：循环仍被堵
`def` 路由 + 阻塞 IO	Starlette 会丢线程池，能跑但占线程
`async def` + 重 CPU	堵循环，应 `run_in_executor`

段末注释：asyncio 是 Python 标准库中的异步 I/O 框架；await 把控制权交还事件循环，等待 IO 完成期间可处理别的请求。

2. 第一性原理：三件事不要混

1
2
3

1. 并发模型：事件循环 / 多 worker / 线程池
2. 语法：async def、await
3. 库：是否真异步（httpx vs requests）

费曼一句：async 不是魔法加速器，只是「等 IO 时别傻站着」的协作方式；傻等同步库仍会站着。

3. 组块：真异步 HTTP 客户端

import httpx
from fastapi import FastAPI

app = FastAPI()


@app.get("/proxy")
async def proxy():
    async with httpx.AsyncClient(timeout=10.0) as client:
        r = await client.get("https://httpbin.org/get")
    return r.json()

阻塞改法对照：

# 坏：在 async 路由里
import requests
r = requests.get("https://httpbin.org/get")  # 阻塞事件循环

# 可接受：改同步路由，让框架丢线程池
@app.get("/proxy-sync")
def proxy_sync():
    return requests.get("https://httpbin.org/get").json()

# 或：run_in_executor 包一层（CPU/遗留同步库）
import asyncio

@app.get("/heavy")
async def heavy():
    return await asyncio.get_event_loop().run_in_executor(None, cpu_bound_fn)

4. BackgroundTasks：响应先回，活稍后干

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_audit_log(message: str):
    with open("audit.log", "a") as f:
        f.write(message + "\n")


@app.post("/orders")
async def create_order(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_audit_log, "order_created")
    return {"status": "accepted"}

维度	BackgroundTasks	独立队列（Celery/ARQ）
可靠性	进程挂了就丢	可持久化、重试
复杂度	极低	需 broker
适用	发邮件、记日志	长任务、分布式

误用：把「必须成功」的支付回调只放 BackgroundTasks——进程重启即丢。

5. 流式响应组块

与 WebSocket 多次推送 的本质差别（一次 HTTP 响应边写边发 vs 双向消息通道）见 01补 §5。

5.1 `StreamingResponse`（通用字节流）

import asyncio

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def event_generator():
    for i in range(5):
        yield f"chunk {i}\n"
        await asyncio.sleep(0.5)


@app.get("/stream")
async def stream():
    return StreamingResponse(event_generator(), media_type="text/plain")

5.2 SSE（Server-Sent Events）

@app.get("/sse")
async def sse():
    async def gen():
        for i in range(5):
            yield f"data: {{\"n\": {i}}}\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(gen(), media_type="text/event-stream")

段末注释：SSE（Server-Sent Events，服务器发送事件）是服务端向浏览器单向推送文本流的 HTTP 机制；双向实时用 WebSocket。

5.3 与 LLM token 流

模式相同：async for chunk in llm_stream(): yield chunk，media_type 按客户端约定（常为 text/event-stream 或 application/x-ndjson）。

6. 辨析交错题

#	代码片段	问题	修法
A	`async def` + `open().read()` 大文件	阻塞	`aiofiles` 或线程池
B	BackgroundTasks 里再 `await` 远程	可以但仍在同一进程	长任务用队列
C	`def` 路由里 `await`	语法错误	改 `async def`
D	StreamingResponse 生成器不 `await` sleep	占循环	生成器用 `async def` + 异步 sleep
E	4 worker 下 BackgroundTasks	每 worker 各自执行	不要指望「全局只跑一次」

7. 合书自测

用一句话说明「假异步」是什么。
BackgroundTasks 与 Celery 各适合什么场景？
StreamingResponse 与一次性 return dict 在 ASGI 层差在哪？（回看 01）

8. 闪卡候选

正面	背面
async 路由里禁止？	同步阻塞 IO（除非线程池）
真异步 HTTP 库示例？	httpx.AsyncClient、aiohttp
BackgroundTasks 执行时机？	响应发送之后、同进程
SSE media_type 常见值？	`text/event-stream`

小结

async 要配合 异步库；阻塞 IO 要么改同步路由+线程池，要么 run_in_executor。
BackgroundTasks 适合轻量后置；流式适合大输出与 LLM token。
原理、Starlette 分流实现与完整踩坑表见 05补异步原理实现与踩坑手册。
下一篇 06 鉴权与安全：API Key / JWT 基线与清单。

1. 问题：async def 写了，为什么还是卡