编程 FastAPI 0.135.x 深度解析:SSE 原生支持、JSON 性能提升 2x+,2026 年 Python Web 框架的性能新标杆

2026-05-14 16:13:28 +0800 CST views 7

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+

核心价值

  1. SSE 原生支持:无需第三方库即可实现服务端推送
  2. JSON 性能提升 2x+:基于 orjson 深度优化
  3. Starlette 1.0+ 兼容:ASGI 工具包成熟稳定
  4. 生态成熟:插件丰富,生产就绪

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.13518,000orjson (最快)✅ ASGI⭐⭐⭐⭐⭐
Flask 3.06,000json (慢)❌ (需插件)⭐⭐⭐⭐
Django 5.08,000json (慢)❌ (需 DRF)⭐⭐⭐
Sanic 23.x15,000ujson (快)✅ ASGI⭐⭐⭐⭐
aiohttp 4.012,000json (慢)✅ ASGI⭐⭐⭐
Tornado 6.09,000json (慢)✅ 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 V2Rust 重写验证速度提升 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
复制全文 生成海报 FastAPI Python Web框架 SSE 性能优化

推荐文章

JavaScript 的模板字符串
2024-11-18 22:44:09 +0800 CST
HTML5的 input:file上传类型控制
2024-11-19 07:29:28 +0800 CST
Golang 几种使用 Channel 的错误姿势
2024-11-19 01:42:18 +0800 CST
LangChain快速上手
2025-03-09 22:30:10 +0800 CST
Vue 中如何处理父子组件通信?
2024-11-17 04:35:13 +0800 CST
Web 端 Office 文件预览工具库
2024-11-18 22:19:16 +0800 CST
CentOS 镜像源配置
2024-11-18 11:28:06 +0800 CST
Vue3中的Store模式有哪些改进?
2024-11-18 11:47:53 +0800 CST
Vue3中的虚拟滚动有哪些改进?
2024-11-18 23:58:18 +0800 CST
Rust 高性能 XML 读写库
2024-11-19 07:50:32 +0800 CST
最全面的 `history` 命令指南
2024-11-18 21:32:45 +0800 CST
使用Vue 3和Axios进行API数据交互
2024-11-18 22:31:21 +0800 CST
Elasticsearch 聚合和分析
2024-11-19 06:44:08 +0800 CST
程序员茄子在线接单