个人博客

主题

FastAPI 与异步 Web 框架 ​

主题：Python 异步编程、FastAPI、WebSocket 与 Web 服务基础设施

1. 总览：为什么 Web 服务需要异步 ​

1.1 Web 服务的典型问题 ​

一个 Web 请求经常不是一直在执行 CPU 计算，而是不断等待外部资源，例如：

  • 数据库返回结果
  • HTTP API 返回结果
  • Redis / 消息系统返回结果
  • 磁盘读写完成
  • 网络数据到达

这类任务通常属于 I/O-bound（I/O 密集型）。

同步模型中，一个请求在等待 I/O 时，当前执行单元通常处于等待状态；当并发请求增加时，就需要更多线程或进程承载请求。

异步模型的核心思想是：

等待 I/O 时主动让出执行权，让事件循环继续处理其他可运行任务；I/O 完成后，再恢复当前协程。

这也是 asyncio 适合网络与 I/O 密集型应用的重要原因。Python 官方文档将 asyncio 定位为使用 async/await 编写并发代码的库，并指出它通常适用于 I/O-bound 与高层网络代码。Python asyncio 官方文档

2. 同步 vs 异步 ​

2.1 同步调用 ​

同步代码执行到某个阻塞 I/O 时，通常需要等操作返回后才能继续：

python

def fetch_data():
    result = sync_client.get("https://example.com")
    return result

可以理解成：

text
开始请求
   ↓
等待 I/O
   ↓
I/O 完成
   ↓
继续执行
   ↓
返回

2.2 异步调用 ​

异步代码将等待点显式表达出来：

python
async def fetch_data():
    result = await async_client.get("https://example.com")
    return result

await 的关键语义不是“更快”，而是：

当前协程暂时无法继续，需要等待一个可等待对象时，把执行权交回事件循环。

Python 官方文档把 coroutine、Task、Future 都归为 awaitable 体系，并说明 Task 可以调度协程并发运行。Python Coroutines and Tasks

2.3 一个关键误区 ​

async def 不会自动把同步代码变成异步代码。

下面的代码虽然使用了 async def，但内部调用仍可能阻塞事件循环：

python
async def handler():
    result = sync_db.query()   # 仍然可能阻塞
    return result

正确思路通常有两种：

  1. 使用原生异步客户端：
python
async def handler():
    result = await async_db.query()
    return result
  1. 无法替换同步库时，把阻塞函数放到线程中执行：
python
import asyncio


def blocking_work():
    ...


async def handler():
    result = await asyncio.to_thread(blocking_work)
    return result

Python 官方文档明确说明，asyncio.to_thread() 用于异步地在线程中运行一个函数，主要面向那些否则会阻塞事件循环的 I/O-bound 函数。Python asyncio.to_thread

3. async def、await、协程与事件循环 ​

3.1 async def ​

python
async def fetch_data():
    ...

async def 定义的是 协程函数。调用它后得到的是 coroutine object（协程对象），并不会因为“调用了函数”就立即执行完整代码。

python
coro = fetch_data()

通常需要通过 await 或调度为 Task 才会真正执行。

3.2 await ​

await 用于等待 awaitable：

python
result = await fetch_data()

常见的 awaitable 包括：

  • coroutine
  • asyncio.Task
  • asyncio.Future

3.3 asyncio.create_task() ​

需要让多个协程并发推进时，可以创建 Task：

python
import asyncio

async def work_a():
    ...

async def work_b():
    ...

async def main():
    task_a = asyncio.create_task(work_a())
    task_b = asyncio.create_task(work_b())

    result_a = await task_a
    result_b = await task_b

这里的“并发”是 concurrency，并不等于 CPU 层面的“并行执行”。FastAPI 官方文档也专门区分 concurrency 与 parallelism，并将异步模型与 I/O 等待场景联系起来。FastAPI：Concurrency and async / await

4. FastAPI 是什么 ​

FastAPI 是面向 Python API 开发的现代 Web 框架，其重要特征包括：

  • 支持 async def / await
  • 基于 Python 类型注解定义参数
  • 使用 Pydantic 做数据建模与校验
  • 自动生成 OpenAPI Schema
  • 提供交互式 API 文档
  • 支持依赖注入
  • 支持中间件
  • 支持 WebSocket
  • 支持应用生命周期管理

FastAPI 官方文档提供 Tutorial、Reference、WebSockets、Lifespan、Middleware、Dependencies 等完整章节。FastAPI 官方学习与参考 · FastAPI Reference

5. FastAPI 最小结构 ​

一个最基本的应用可以抽象为：

python
from fastapi import FastAPI

app = FastAPI()


@app.get("/hello")
async def hello():
    return {"message": "hello"}

核心概念有三个：

概念含义
FastAPI()创建应用对象
@app.get(...)声明 HTTP 路由
路由函数接收请求并产生响应

FastAPI 会根据路由声明和类型注解构建 OpenAPI Schema，并据此生成交互式文档。

6. 路由（Path Operation） ​

6.1 路由的本质 ​

HTTP API 可以理解为：

text
HTTP Method + Path + Handler

例如：

python
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

FastAPI 会结合：

  • 路径参数
  • 查询参数
  • 请求体
  • 类型注解

来决定如何解析请求数据。

6.2 常见 HTTP 方法 ​

方法常见语义
GET获取资源
POST创建或提交数据
PUT整体更新
PATCH部分更新
DELETE删除资源

7. def 与 async def 的选择 ​

这是 FastAPI 使用异步模型时最重要的边界之一。

7.1 可以使用 async def 的情况 ​

当调用的库本身支持异步接口：

python
@app.get("/data")
async def get_data():
    result = await async_client.fetch()
    return result

7.2 可以使用普通 def 的情况 ​

当主要依赖同步库，尤其存在阻塞 I/O 时，可以使用普通 def。

FastAPI 官方文档说明：对于普通 def 的路径操作函数，FastAPI 会将其放到外部线程池执行，而不是直接在事件循环中调用。FastAPI async 文档

python
@app.get("/data")
def get_data():
    return sync_client.fetch()

7.3 async def 中直接调用同步阻塞函数 ​

需要特别避免：

python
@app.get("/bad")
async def bad():
    return sync_client.fetch()  # 可能阻塞事件循环

如果必须在异步上下文中调用同步阻塞函数，可以显式使用：

python
result = await asyncio.to_thread(sync_client.fetch)

7.4 判断原则 ​

可以简单记成：

text
库支持 await
    → async def + await

库只有同步接口
    → 普通 def
    或
    async def + asyncio.to_thread(...)

FastAPI 官方文档还说明，依赖函数也遵循类似规则：普通 def 依赖可以在线程池中运行，async def 依赖则直接以异步方式运行。FastAPI Dependencies

8. Pydantic：请求数据模型与校验 ​

FastAPI 使用 Pydantic 模型定义结构化请求数据。

python
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    price: float
    description: str | None = None

然后可以直接作为路径操作函数参数：

python
@app.post("/items")
async def create_item(item: Item):
    return item

FastAPI 会根据类型声明完成一系列工作：

  1. 读取请求体
  2. 将输入数据解析为模型
  3. 进行数据校验
  4. 将解析后的对象传给函数
  5. 生成相应 JSON Schema / OpenAPI 信息

官方文档明确指出，Pydantic 模型声明不仅承担校验，也会进入生成的 OpenAPI Schema 和交互式文档。FastAPI Request Body

8.1 Field ​

可以使用 Field 增加约束和元数据：

python
from pydantic import BaseModel, Field


class Item(BaseModel):
    name: str = Field(min_length=1)
    price: float = Field(gt=0)

这样，数据模型本身就成为 API 契约的一部分。FastAPI Body Fields

9. APIRouter：路由模块化 ​

当 API 数量增加时，不应把所有路径操作都堆在一个文件里。

FastAPI 提供 APIRouter 进行模块化拆分：

python
from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])


@router.get("/")
async def list_users():
    return []

再把 Router 注册到主应用：

python
app.include_router(router)

可以把路由按职责拆分成多个模块，例如：

text
app
├── main.py
├── routers/
│   ├── users.py
│   ├── items.py
│   └── auth.py
└── models/

APIRouter 的价值主要是：

  • 代码组织
  • 公共前缀
  • tags
  • 公共依赖
  • 模块边界
  • API 文档结构

官方 Reference 也将 APIRouter 作为核心 API 组件，并支持通过 lifespan、include_in_schema 等参数配置路由行为。FastAPI APIRouter Reference

10. 依赖注入（Dependency Injection） ​

FastAPI 的依赖注入常通过 Depends 使用。

python
from fastapi import Depends


def get_current_user():
    return ...


@app.get("/profile")
async def profile(user=Depends(get_current_user)):
    return user

可以把依赖理解成：

text
请求
 ↓
解析依赖
 ↓
执行依赖函数
 ↓
得到结果
 ↓
注入到路由函数
 ↓
执行路由

典型用途：

  • 鉴权
  • 数据库会话
  • 公共查询参数
  • 权限检查
  • 配置或共享资源
  • 可复用的请求前置逻辑

FastAPI 官方文档强调：依赖可以继续包含子依赖，并且依赖的参数、校验和要求可以进入 OpenAPI Schema。FastAPI Dependencies

10.1 依赖也有异步边界 ​

python
async def get_resource():
    return await async_client.fetch()

或者：

python
def get_resource():
    return sync_client.fetch()

两者都可以，关键是让依赖内部的调用方式和函数定义保持一致。

11. Middleware：中间件 ​

中间件位于请求处理链的外围。

可以抽象为：

text
请求
 ↓
Middleware
 ↓
路由
 ↓
业务处理
 ↓
Middleware
 ↓
响应

FastAPI 官方定义中间件为：

在某个具体路径操作处理之前接收请求，并在响应返回之前再次获得控制权。

因此它适合做跨请求的统一处理，例如：

  • CORS
  • 日志
  • 请求耗时统计
  • Trace / Request ID
  • 通用响应头
  • 某些统一安全策略

基本结构：

python
@app.middleware("http")
async def timing_middleware(request, call_next):
    response = await call_next(request)
    return response

官方文档还说明了多个中间件的包装顺序：后添加的中间件处于更外层，请求方向从外向内，响应方向从内向外。FastAPI Middleware

12. CORS ​

CORS（Cross-Origin Resource Sharing，跨域资源共享）是浏览器的安全机制之一。

当前端页面和 API 位于不同 origin 时，浏览器会根据 CORS 规则判断请求是否允许。

FastAPI 可以直接通过 CORSMiddleware 配置：

python
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

12.1 配置注意点 ​

生产环境不建议无条件开放所有来源，应根据真实部署场景配置 allow_origins。

官方文档将 CORS 作为标准中间件能力提供，并给出了 CORSMiddleware 的参数定义。FastAPI Middleware Reference

13. 应用生命周期（Lifespan） ​

有些资源不是“每个请求创建一次”，而是希望：

text
应用启动
  ↓
初始化共享资源
  ↓
开始接收请求
  ↓
持续运行
  ↓
停止接收请求
  ↓
释放资源
  ↓
应用退出

FastAPI 推荐通过 lifespan 统一管理启动与关闭逻辑。

典型形式：

python
from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    resource = create_resource()
    yield
    # shutdown
    resource.close()


app = FastAPI(lifespan=lifespan)

yield 前可以做启动初始化，yield 后可以做清理。

适合管理的资源包括：

  • 数据库连接池
  • HTTP 客户端
  • 缓存客户端
  • 共享模型
  • 文件或网络资源
  • 其他跨请求共享的对象

FastAPI 官方文档说明，lifespan 逻辑在应用开始接收请求前执行启动阶段，在应用结束前执行清理阶段；当前官方推荐使用 lifespan，而不是旧式 startup/shutdown event handlers。FastAPI Lifespan Events

14. WebSocket ​

14.1 HTTP 的基本模型 ​

HTTP 常见交互方式可以简化为：

text
客户端 → 请求 → 服务端
客户端 ← 响应 ← 服务端

它天然适合一次请求、一次响应的 API 场景。

14.2 WebSocket 的基本模型 ​

WebSocket 建立连接后，可以在同一连接上进行双向通信：

text
客户端 ⇄ WebSocket 连接 ⇄ 服务端

服务端可以在没有收到新的 HTTP 请求时主动发送消息。

MDN 将 WebSocket 描述为在浏览器和服务器之间开启双向交互式通信会话，而不需要客户端持续轮询服务器等待回复。MDN WebSocket API

15. FastAPI 中的 WebSocket ​

FastAPI 提供 WebSocket 支持。

一个最小结构：

python
from fastapi import FastAPI, WebSocket

app = FastAPI()


@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    while True:
        data = await websocket.receive_text()
        await websocket.send_text(data)

核心步骤：

  1. 接受连接 accept()
  2. 接收消息 receive_text() / receive_json() 等
  3. 处理消息
  4. 发送 send_text() / send_json() 等
  5. 连接关闭时结束处理

FastAPI 官方 WebSocket 文档展示了上述基本模式，并说明 WebSocket 路由可以使用 await 等待消息与发送消息。FastAPI WebSockets

16. HTTP vs WebSocket ​

特性HTTPWebSocket
典型模式请求-响应双向持续通信
服务端主动推送通常需要额外机制原生支持
长连接交互可以借助其他技术实现原生适合
典型场景CRUD、普通 API实时消息、实时状态、交互式流
数据流方向以请求-响应为主客户端与服务端都可主动发送

选择时不要简单理解成“WebSocket 比 HTTP 更高级”，而应该看通信模式：

一次性数据获取倾向 HTTP；持续双向交互倾向 WebSocket（例如逐token返回大模型答案）。

17. WebSocket 与流式输出 ​

对于需要持续生成结果的场景，流式通信的关键不是“必须使用 WebSocket”，而是：

不要等所有结果都生成完成后再一次性返回，而是让客户端持续收到增量数据。

例如可以抽象成：

text
开始
 ↓
阶段性状态
 ↓
增量数据 1
 ↓
增量数据 2
 ↓
增量数据 3
 ↓
结束

WebSocket 非常适合这种模式，因为服务器可以在一个连接上持续发送多个消息。FastAPI WebSockets · MDN WebSocket API

18. Generator 与流式处理 ​

Python Generator 使用 yield 逐步产出值：

python
def stream_data():
    yield "part-1"
    yield "part-2"
    yield "part-3"

消费方可以逐个取得结果：

python
for item in stream_data():
    print(item)

在 Web 服务中，它适合表达：

text
产生一个结果
 ↓
发送
 ↓
继续计算
 ↓
再产生一个结果
 ↓
再发送

18.1 同步 Generator + 异步 WebSocket ​

如果生成器内部是同步阻塞操作，而外层是异步 WebSocket Handler，可以通过 asyncio.to_thread() 让“推进生成器”的动作在线程中执行：

python
async def websocket_endpoint(websocket):
    await websocket.accept()

    generator = stream_data()

    while True:
        has_value, value = await asyncio.to_thread(next_item, generator)

        if not has_value:
            break

        await websocket.send_text(value)

这种模式的核心是：

text
WebSocket Handler
      │
      ├── await receive()
      │
      ├── await to_thread(sync_step)
      │
      └── await send()

也就是说：WebSocket 层保持异步，内部同步阻塞步骤显式隔离。

19. 事件循环中最容易踩的坑 ​

19.1 在 async 函数里直接调用阻塞 I/O ​

python
async def handler():
    time.sleep(5)    # 错误示例：阻塞事件循环

应该优先使用：

python
await asyncio.sleep(5)

如果确实是阻塞函数：

python
await asyncio.to_thread(blocking_func)

19.2 “使用 async”不等于“已经并发” ​

python
async def handler():
    await task_a()
    await task_b()

这里通常是先等待 task_a，再等待 task_b。

如果两个操作彼此独立，需要并发推进，可以考虑 Task 或其他 asyncio 并发原语。

python
async def handler():
    task_a = asyncio.create_task(task_a())
    task_b = asyncio.create_task(task_b())

    result_a, result_b = await task_a, await task_b

19.3 混用同步客户端与异步客户端 ​

一个函数内部应该明确：

text
这一步到底是否可以 await？

如果库接口是同步的，就不能因为外层写了 async def 而假设它不会阻塞。

20. 异步框架的性能理解 ​

不要把异步简单理解成：

“异步 = 单线程，所以一定更快。”

更准确的理解是：

text
I/O 等待很多
    ↓
大量时间花在等待外部系统
    ↓
事件循环可以在等待期间处理其他任务
    ↓
提高单个进程对并发 I/O 请求的利用率

所以异步尤其适合：

  • 网络 API
  • 数据库访问
  • 高并发 I/O
  • WebSocket
  • 流式通信

而对于纯 CPU 密集型任务，单纯增加 async/await 并不会自动带来性能提升。

21. 一个完整的 FastAPI 请求处理思维模型 ​

可以把一次 HTTP 请求理解成：

text
客户端请求
   ↓
ASGI Server
   ↓
Middleware
   ↓
路由匹配
   ↓
参数解析 / Pydantic 校验
   ↓
Dependency Injection
   ↓
Path Operation Function
   ↓
业务代码
   ↓
返回值序列化
   ↓
Middleware
   ↓
HTTP Response

而 WebSocket 可以理解成：

text
客户端发起 WebSocket 连接
   ↓
WebSocket Handler
   ↓
accept()
   ↓
持续 receive()
   ↓
处理 / await / to_thread
   ↓
send()
   ↓
持续通信
   ↓
断开连接

22. 核心概念速记 ​

异步编程 ​

  • async def：定义协程函数
  • await：等待 awaitable，并把执行权交回事件循环
  • asyncio.create_task()：调度协程并发运行
  • asyncio.to_thread()：把阻塞函数放到线程中执行
  • 异步最适合 I/O-bound 场景
  • async def 不会自动消除内部同步阻塞

FastAPI ​

  • FastAPI()：应用对象
  • @app.get() / @app.post()：声明路径操作
  • APIRouter：模块化路由
  • Pydantic：请求/响应数据模型与校验
  • Depends()：依赖注入
  • Middleware：请求/响应外围统一处理
  • lifespan：应用启动与关闭资源管理
  • OpenAPI：自动生成 API Schema 与交互式文档

WebSocket ​

  • 长连接
  • 双向通信
  • 服务端可主动推送
  • 适合实时通信与增量流式输出
  • FastAPI 中通常使用异步 WebSocket Handler
  • 断开连接属于正常生命周期事件，应做好异常处理

23. 常见面试 / 复习问题 ​

Q1：为什么异步适合 Web 服务？ ​

因为 Web 服务大量时间花在等待 I/O。异步可以在一个任务等待期间，让事件循环推进其他任务，从而提高并发 I/O 的处理效率。

Q2：async def 是不是就不会阻塞？ ​

不是。只要内部直接调用阻塞式函数，就仍可能阻塞事件循环。需要使用原生异步库，或者把同步阻塞工作移到线程中。

Q3：什么时候用 def，什么时候用 async def？ ​

能自然 await 的异步库调用 → async def；主要使用同步阻塞库时 → 可使用普通 def，或者在确有需要时使用 asyncio.to_thread()。

Q4：FastAPI 为什么需要 Pydantic？ ​

因为它把输入数据的结构、类型和约束显式化，同时支持运行时校验、类型提示和 OpenAPI Schema 生成。

Q5：APIRouter 是干什么的？ ​

主要用于把大量路由拆成多个逻辑模块，并统一管理前缀、标签和依赖等配置。

Q6：Middleware 和 Dependency 有什么区别？ ​

可以粗略理解：

  • Middleware：站在整个请求/响应链外围
  • Dependency：作为具体路径操作的依赖项参与执行

两者都能处理通用逻辑，但作用范围和组织方式不同。

Q7：lifespan 为什么重要？ ​

因为数据库连接池、共享客户端、模型等资源具有“启动一次、运行期间复用、关闭时清理”的生命周期，lifespan 能把这类逻辑集中管理。

Q8：WebSocket 与 HTTP 最大区别？ ​

HTTP 的典型模式是请求-响应；WebSocket 建立连接后可以长期保持双向通信，服务器可以主动向客户端发送消息。

24. 最后形成一条知识链 ​

text
Python async/await
        ↓
     asyncio
        ↓
   Event Loop
        ↓
I/O 等待时让出执行权
        ↓
异步 Web 框架 / ASGI
        ↓
     FastAPI
        ↓
路由 + Pydantic + Depends + Middleware
        ↓
     Lifespan
        ↓
   WebSocket
        ↓
持续双向通信 / 流式输出

最核心的一句话：

异步 Web 的关键不在于“所有代码都写成 async def”，而在于正确区分可等待的 I/O、同步阻塞操作与 CPU 密集型任务，并让阻塞操作不要把事件循环卡住。

25. 参考文档 ​

Python 官方文档 ​

  1. asyncio — Asynchronous I/O
  2. Coroutines and Tasks
  3. asyncio.to_thread()

FastAPI 官方文档 ​

  1. Concurrency and async / await
  2. Learn FastAPI
  3. Request Body
  4. Dependencies
  5. Middleware
  6. Lifespan Events
  7. WebSockets
  8. APIRouter Reference

Web API 文档 ​

26. 一页速查表 ​

关键词一句话理解
I/O-bound大量时间花在等待外部资源
async声明协程函数
await等待并让出执行权
Event Loop调度异步任务的核心机制
Task被调度执行的协程
to_thread把阻塞函数放入线程执行
FastAPIPython API Web 框架
Path OperationHTTP 路由处理函数
APIRouter路由模块化
Pydantic数据模型与运行时校验
Depends依赖注入
Middleware请求/响应外围统一处理
CORS浏览器跨域访问控制机制
lifespan应用启动与关闭生命周期管理
WebSocket双向持续通信协议
Generator通过 yield 持续产生数据
Streaming增量地产生并发送结果