FastAPI 0.135.x 深度解析:SSE 原生支持、JSON 性能提升 2x+,2026 年 Python Web 框架的性能新标杆
FastAPI 0.135.x 于 2026 年 3 月初发布,支持 Starlette 1.0+,新增 Server-Sent Events(SSE)原生支持,JSON 响应性能提升 2x+,整体生态非常成熟。本文深度解析 FastAPI 0.135.x 的核心新特性、SSE 实时推送实战、JSON 性能优化原理、Starlette 1.0 带来的架构升级,以及与 Flask、Django、Sanic 的横向对比。
一、FastAPI 0.135.x 版本定位
1.1 为什么 0.135.x 是 2026 年最重要的版本
FastAPI 版本演进(2024-2026):
0.104.x (2024 Q2) ───→ 异步基础、Pydantic V2 支持
0.110.x (2024 Q4) ───→ OpenAPI 3.1、WebSocket 增强
0.115.x (2025 Q2) ───→ 依赖注入优化、Lifespan 事件
0.135.x (2026 Q1) ───→ ★ SSE 原生支持、JSON 性能 2x、Starlette 1.0+
核心价值:
- SSE 原生支持:无需第三方库即可实现服务端推送
- JSON 性能提升 2x+:基于
orjson深度优化 - Starlette 1.0+ 兼容:ASGI 工具包成熟稳定
- 生态成熟:插件丰富,生产就绪
1.2 升级检查
# 检查当前版本
python -c "import fastapi; print(fastapi.__version__)"
# 升级到 0.135.x
pip install -U "fastapi>=0.135.0"
# 同时升级 Starlette
pip install -U "starlette>=1.0.0"
# 验证
python -c "import starlette; print(starlette.__version__)"
二、SSE(Server-Sent Events)原生支持
2.1 SSE 是什么,为什么需要它
HTTP 轮询(旧方案):
Client ─── GET /api/status ──→ Server
Client <── {"status": "pending"} ─── Server
(等待 3 秒)
Client ─── GET /api/status ──→ Server
Client <── {"status": "running"} ─── Server
(等待 3 秒)
Client ─── GET /api/status ──→ Server
Client <── {"status": "done"} ─── Server
❌ 延迟高、服务器压力大、浪费带宽
SSE(Server-Sent Events):
Client ─── GET /api/status ──→ Server
Client <── "pending" ─── Server(保持连接)
Client <── "running" ─── Server(推送)
Client <── "done" ─── Server(推送)
✅ 低延迟、单向流式推送、自动重连
2.2 FastAPI 0.135.x 的 SSE 支持
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
# 方式 1:使用 StreamingResponse(兼容旧版)
async def event_generator():
"""生成 SSE 事件流"""
for i in range(10):
# SSE 格式:data: <content>\n\n
data = {"progress": i * 10, "status": "processing"}
yield f"data: {json.dumps(data)}\n\n"
await asyncio.sleep(1)
# 结束事件
yield f"data: {json.dumps({'progress': 100, 'status': 'done'})}\n\n"
@app.get("/api/progress")
async def get_progress():
return StreamingResponse(
event_generator(),
media_type="text/event-stream"
)
# 方式 2:FastAPI 0.135.x 原生 SSE 支持(推荐)
from fastapi.responses import EventSourceResponse
async def sse_event_generator():
"""SSE 事件生成器(0.135.x 优化版)"""
for i in range(10):
# 直接返回字典,FastAPI 自动序列化为 SSE 格式
yield {"event": "progress", "data": {"percent": i * 10}}
await asyncio.sleep(1)
yield {"event": "done", "data": {"result": "任务完成"}}
@app.get("/api/progress/v2")
async def get_progress_v2():
return EventSourceResponse(sse_event_generator())
# 前端接收(JavaScript)
const eventSource = new EventSource('/api/progress/v2');
eventSource.addEventListener('progress', (event) => {
const data = JSON.parse(event.data);
console.log(`进度: ${data.percent}%`);
});
eventSource.addEventListener('done', (event) => {
const data = JSON.parse(event.data);
console.log(data.result);
eventSource.close();
});
2.3 实战:AI 流式回答
from fastapi import FastAPI
from fastapi.responses import EventSourceResponse
import asyncio
import json
app = FastAPI()
# 模拟 AI 流式生成
async def ai_stream_generator(prompt: str):
"""模拟 AI 逐 token 生成"""
response = f"您的问题是:{prompt}。让我来回答..."
# 逐字推送
for char in response:
yield {
"event": "token",
"data": {"token": char}
}
await asyncio.sleep(0.05) # 模拟生成延迟
# 结束事件
yield {
"event": "done",
"data": {"full_text": response}
}
@app.get("/api/ai/chat")
async def ai_chat(prompt: str):
return EventSourceResponse(ai_stream_generator(prompt))
# 前端接收
const eventSource = new EventSource(`/api/ai/chat?prompt=${encodeURIComponent("FastAPI 是什么?")}`);
let fullText = "";
eventSource.addEventListener('token', (event) => {
const data = JSON.parse(event.data);
fullText += data.token;
document.getElementById('answer').innerText = fullText;
});
eventSource.addEventListener('done', (event) => {
const data = JSON.parse(event.data);
console.log('完整回答:', data.full_text);
eventSource.close();
});
三、JSON 性能提升 2x+ 的原理
3.1 问题:为什么 JSON 序列化慢
# 传统 json 模块(慢)
import json
from pydantic import BaseModel
from fastapi import FastAPI
app = FastAPI()
class User(BaseModel):
id: int
name: str
email: str
# ... 20 个字段
# FastAPI 默认使用 json.dumps()
# 对于复杂对象,序列化耗时
# 基准测试
import time
users = [User(id=i, name=f"User{i}", email=f"user{i}@example.com") for i in range(10000)]
start = time.time()
json_data = json.dumps([u.dict() for u in users])
end = time.time()
print(f"json.dumps: {end - start:.3f}s") # 约 0.8 秒
3.2 FastAPI 0.135.x 的 JSON 优化
FastAPI 0.135.x JSON 性能优化:
1. 默认使用 orjson(如果已安装)
- orjson 比 json 快 2-3 倍
- 支持 dataclass、NamedTuple、datetime 等类型
- 零拷贝序列化
2. 自动选择最优 JSON 库
- 优先 orjson
- 其次 ujson
- 最后标准库 json
3. Pydantic V2 深度集成
- 基于 Rust 重写的 Pydantic V2
- 模型序列化性能提升 5-10 倍
3.3 性能对比
# 安装 orjson(推荐)
pip install orjson
# FastAPI 自动检测并使用 orjson
from fastapi import FastAPI
from pydantic import BaseModel
import json
import orjson
import time
app = FastAPI()
class User(BaseModel):
id: int
name: str
email: str
age: int
is_active: bool
tags: list[str]
metadata: dict
# 生成测试数据
users = [
User(
id=i,
name=f"User{i}",
email=f"user{i}@example.com",
age=20 + i % 50,
is_active=i % 2 == 0,
tags=[f"tag{j}" for j in range(5)],
metadata={"score": i * 1.5, "rank": i % 100}
)
for i in range(10000)
]
# 基准测试 1:标准库 json
def serialize_json(users):
return json.dumps([u.dict() for u in users])
# 基准测试 2:orjson
def serialize_orjson(users):
return orjson.dumps([u.model_dump() for u in users])
# 基准测试 3:Pydantic V2 model_dump_json()
def serialize_pydantic_v2(users):
return [u.model_dump_json() for u in users]
# 运行测试
iterations = 100
# json.dumps
start = time.time()
for _ in range(iterations):
serialize_json(users)
json_time = time.time() - start
# orjson
start = time.time()
for _ in range(iterations):
serialize_orjson(users)
orjson_time = time.time() - start
# Pydantic V2
start = time.time()
for _ in range(iterations):
serialize_pydantic_v2(users)
pydantic_time = time.time() - start
print(f"json.dumps: {json_time:.2f}s")
print(f"orjson: {orjson_time:.2f}s ({(json_time/orjson_time):.1f}x faster)")
print(f"Pydantic V2: {pydantic_time:.2f}s ({(json_time/pydantic_time):.1f}x faster)")
# 输出示例:
# json.dumps: 5.20s
# orjson: 1.80s (2.9x faster)
# Pydantic V2: 1.20s (4.3x faster)
3.4 强制使用 orjson
from fastapi import FastAPI
from fastapi.responses import JSONResponse
import orjson
app = FastAPI()
# 自定义 JSONResponse,强制使用 orjson
class OrjsonResponse(JSONResponse):
def render(self, content) -> bytes:
return orjson.dumps(content)
@app.get("/api/users", response_class=OrjsonResponse)
async def get_users():
users = [{"id": i, "name": f"User{i}"} for i in range(1000)]
return users # 自动使用 orjson 序列化
四、Starlette 1.0+ 带来的架构升级
4.1 Starlette 是什么
FastAPI 架构层次:
┌─────────────────────────────────────────┐
│ FastAPI(你的代码) │
│ - 路由装饰器 @app.get() │
│ - 依赖注入 Depends() │
│ - 数据验证 Pydantic │
│ - OpenAPI 文档自动生成 │
└──────────────┬──────────────────────────┘
│
┌───────────────┴──────────────────────────┐
│ Starlette 1.0+(ASGI 工具包) │
│ - 路由匹配 │
│ - 请求/响应封装 │
│ - WebSocket 支持 │
│ - 静态文件服务 │
│ - 测试客户端 TestClient │
└──────────────┬──────────────────────────┘
│
┌───────────────┴──────────────────────────┐
│ Uvicorn / Hypercorn(ASGI 服务器)│
│ - 异步 I/O 事件循环 │
│ - HTTP/1.1、HTTP/2、WebSocket │
└─────────────────────────────────────────┘
4.2 Starlette 1.0 新特性
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
# Starlette 1.0 新特性:
# 1. 更简洁的路由定义
async def homepage(request):
return JSONResponse({"hello": "world"})
app = Starlette(debug=True, routes=[
Route('/', homepage),
])
# 2. 中间件支持异步
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
import time
start = time.time()
response = await call_next(request)
end = time.time()
response.headers['X-Process-Time'] = str(end - start)
return response
app = Starlette(
routes=[...],
middleware=[
Middleware(TimingMiddleware),
Middleware(CORSMiddleware, allow_origins=['*'])
]
)
# 3. WebSocket 增强
from starlette.websockets import WebSocket
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_json()
await websocket.send_json({"echo": data})
# 4. 静态文件服务增强
from starlette.staticfiles import StaticFiles
app.mount('/static', StaticFiles(directory='static'), name='static')
4.3 FastAPI + Starlette 1.0 整合
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
app = FastAPI()
# FastAPI 0.135.x 自动继承 Starlette 1.0 的能力
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["example.com", "*.example.com"]
)
五、与其他框架的横向对比(2026 年现状)
5.1 性能对比
| 框架 | 请求/秒(单核) | JSON 序列化 | 异步原生 | 易用性 |
|---|---|---|---|---|
| FastAPI 0.135 | 18,000 | orjson (最快) | ✅ ASGI | ⭐⭐⭐⭐⭐ |
| Flask 3.0 | 6,000 | json (慢) | ❌ (需插件) | ⭐⭐⭐⭐ |
| Django 5.0 | 8,000 | json (慢) | ❌ (需 DRF) | ⭐⭐⭐ |
| Sanic 23.x | 15,000 | ujson (快) | ✅ ASGI | ⭐⭐⭐⭐ |
| aiohttp 4.0 | 12,000 | json (慢) | ✅ ASGI | ⭐⭐⭐ |
| Tornado 6.0 | 9,000 | json (慢) | ✅ ASGI | ⭐⭐⭐ |
5.2 特性对比
特性对比表(2026 年):
┌──────────────────┬──────────┬─────────┬─────────┬─────────┐
│ 特性 │ FastAPI │ Flask │ Django │ Sanic │
├──────────────────┼──────────┼─────────┼─────────┼─────────┤
│ 异步原生 │ ✅ ASGI │ ❌ │ △ DRF │ ✅ ASGI │
│ 数据验证 │ ✅ Pydantic│ ❌ │ △ DRF │ △ │
│ OpenAPI 文档 │ ✅ 自动 │ ❌ │ △ DRF │ △ │
│ SSE 支持 │ ✅ 原生 │ △ 手动 │ △ 手动 │ ✅ 原生 │
│ JSON 性能 │ ⭐⭐⭐⭐⭐ │ ⭐⭐ │ ⭐⭐ │ ⭐⭐⭐⭐ │
│ WebSocket │ ✅ │ △ 插件 │ △ DRF │ ✅ │
│ 依赖注入 │ ✅ │ ❌ │ △ │ △ │
│ 生产成熟度 │ ⭐⭐⭐⭐ │ ⭐⭐⭐⭐⭐│ ⭐⭐⭐⭐⭐│ ⭐⭐⭐ │
│ 学习曲线 │ 中等 │ 简单 │ 中等 │ 中等 │
└──────────────────┴──────────┴─────────┴─────────┴─────────┘
5.3 选型建议
✅ 选择 FastAPI 0.135.x 的场景:
1. 需要高性能 JSON API
2. 需要异步支持(高并发)
3. 需要自动生成 API 文档
4. 需要数据验证(Pydantic)
5. 需要 SSE / WebSocket 实时通信
✅ 选择 Flask 的场景:
1. 小型项目、原型开发
2. 不需要异步
3. 团队熟悉 Flask
✅ 选择 Django 的场景:
1. 需要完整的后台管理系统
2. 需要 ORM(Django ORM)
3. 快速开发 CRUD 应用
✅ 选择 Sanic 的场景:
1. 需要异步 + 高性能
2. 不需要 Pydantic 数据验证
六、完整项目实战:RESTful API + SSE 进度推送
6.1 项目结构
fastapi_0135_demo/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 应用入口
│ ├── api/
│ │ ├── __init__.py
│ │ ├── users.py # 用户 API
│ │ └── tasks.py # 任务 API(含 SSE)
│ ├── models/
│ │ ├── __init__.py
│ │ └── user.py # Pydantic 模型
│ └── core/
│ ├── __init__.py
│ └── config.py # 配置
├── tests/
│ └── test_api.py
├── requirements.txt
└── run.py
6.2 完整代码
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import orjson
# 生命周期事件(替代 on_event)
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时执行
print("应用启动")
yield
# 关闭时执行
print("应用关闭")
app = FastAPI(
title="FastAPI 0.135 Demo",
description="SSE 原生支持 + JSON 性能优化演示",
version="0.135.0",
lifespan=lifespan
)
# CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 路由
from app.api import users, tasks
app.include_router(users.router, prefix="/api/users", tags=["用户管理"])
app.include_router(tasks.router, prefix="/api/tasks", tags=["任务管理"])
# 健康检查
@app.get("/health")
async def health_check():
return {"status": "ok", "version": "0.135.0"}
# app/api/tasks.py(SSE 任务进度推送)
from fastapi import APIRouter, HTTPException
from fastapi.responses import EventSourceResponse
from pydantic import BaseModel
import asyncio
import uuid
router = APIRouter()
# 任务模型
class TaskCreate(BaseModel):
name: str
description: str = ""
# 模拟任务存储
tasks_db = {}
# SSE 事件生成器
async def task_progress_generator(task_id: str):
"""模拟任务执行,推送进度"""
steps = [
"初始化任务...",
"数据预处理...",
"模型推理中...",
"后处理...",
"保存结果..."
]
for i, step in enumerate(steps):
# 推送进度
yield {
"event": "progress",
"data": {
"task_id": task_id,
"percent": (i + 1) * 20,
"step": step
}
}
await asyncio.sleep(1) # 模拟处理延迟
# 推送完成事件
result = {"task_id": task_id, "status": "success", "result": "任务完成"}
yield {
"event": "done",
"data": result
}
# 更新任务状态
tasks_db[task_id] = result
@router.post("/")
async def create_task(task: TaskCreate):
"""创建任务"""
task_id = str(uuid.uuid4())
tasks_db[task_id] = {"status": "pending", "name": task.name}
return {"task_id": task_id, "status": "pending"}
@router.get("/{task_id}/progress")
async def get_task_progress(task_id: str):
"""获取任务进度(SSE)"""
if task_id not in tasks_db:
raise HTTPException(status_code=404, detail="Task not found")
return EventSourceResponse(task_progress_generator(task_id))
# tests/test_api.py
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_health_check():
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
def test_create_task():
response = client.post("/api/tasks/", json={
"name": "测试任务",
"description": "这是一个测试任务"
})
assert response.status_code == 200
data = response.json()
assert "task_id" in data
assert data["status"] == "pending"
def test_sse_progress():
# 创建任务
response = client.post("/api/tasks/", json={"name": "SSE 测试"})
task_id = response.json()["task_id"]
# 获取 SSE 流
response = client.get(f"/api/tasks/{task_id}/progress")
assert response.status_code == 200
assert "text/event-stream" in response.headers["content-type"]
七、生产部署指南
7.1 使用 Uvicorn 部署
# 安装 Gunicorn + Uvicorn worker
pip install gunicorn uvicorn
# 启动(生产模式)
gunicorn app.main:app \
--workers 4 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--access-logfile - \
--error-logfile -
# 使用 Uvicorn 直接启动(开发模式)
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
7.2 Docker 部署
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 复制代码
COPY . .
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["gunicorn", "app.main:app", \
"--workers", "4", \
"--worker-class", "uvicorn.workers.UvicornWorker", \
"--bind", "0.0.0.0:8000"]
# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- PYTHONUNBUFFERED=1
volumes:
- .:/app
command: gunicorn app.main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
八、总结
8.1 FastAPI 0.135.x 核心新特性
| 特性 | 说明 | 性能提升 |
|---|---|---|
| SSE 原生支持 | 无需第三方库 | 实时推送延迟 < 50ms |
| JSON 性能优化 | 默认 orjson | 序列化速度提升 2-3x |
| Starlette 1.0+ | ASGI 工具包成熟 | 路由性能提升 20% |
| Pydantic V2 | Rust 重写 | 验证速度提升 5-10x |
| Lifespan 事件 | 替代 on_event | 更清晰的启动/关闭逻辑 |
8.2 升级建议
✅ 推荐升级到 0.135.x 的场景:
1. 需要 SSE 实时推送
2. JSON API 性能瓶颈
3. 需要异步高并发
4. 新项目(直接享受新特性)
⚠️ 升级前注意:
1. 检查 Starlette 版本兼容性
2. 检查 Pydantic V2 迁移(如果是从 V1 升级)
3. 运行完整测试套件
一句话总结:FastAPI 0.135.x 是 2026 年 Python Web 框架的性能新标杆——SSE 原生支持让实时推送变得简单,JSON 性能提升 2x+ 让 API 响应更快,Starlette 1.0+ 让 ASGI 架构更加成熟。如果你在用 Python 做 Web 开发,这是目前最好的选择。
参考资源:
- FastAPI 官方文档:https://fastapi.tiangolo.com/
- FastAPI 0.135 发布说明:https://github.com/tiangolo/fastapi/releases/tag/0.135.0
- Starlette 1.0 文档:https://www.starlette.io/
- Pydantic V2 迁移指南:https://docs.pydantic.dev/latest/migration/
- orjson GitHub:https://github.com/ijl/orjson