从Flask跳到FastAPI后我的接口响应时间少了80%

阳光_思想家
2026-07-01 07:32
阅读 713

作者:一个曾经死磕VSCode插件、现在却天天跟AI抢键盘的前端转全栈打工人


先说个事儿。

上周五晚上十点半,我坐在工位上,盯着屏幕上那个转了快三秒的loading圈,脑子里只有一个念头:这破接口到底在干嘛?

坐标北京,刚挤完一号线回到公司(别问为什么十点还在,问就是需求变更)。产品经理下午临时加了个数据聚合的需求,我用Flask吭哧吭哧写完,一跑——好家伙,P99延迟直接飙到2800ms。测试小姐姐在旁边幽幽地来了句:"这用户体验,用户怕是要把手机扔了。"

我当时真的想砸键盘。

后来我花了个周末把后端从Flask迁移到了FastAPI,同样的逻辑,同样的数据量,P99延迟降到了400ms以内。没错,少了80%。

今天就来聊聊FastAPI这个让我"真香"的框架。顺便也聊聊,一个写前端动画的,是怎么一步步被"逼"上后端这条不归路的。

一个前端仔的后端血泪史

先交代下背景。

我在北京一家做SaaS的公司,主要搞前端,平时跟Vue3、GSAP、Lottie打交道比较多。但你知道的,小公司嘛,"全栈"两个字就像紧箍咒,后端没人写的时候,那活儿就自然落到你头上了。

我之前写后端,用的是Flask。不是说Flask不好,它确实轻量、灵活,插件生态也丰富。我VSCode里光Python相关的插件就装了十几个——Python Extension Pack、Pylance、Black Formatter、Ruff、even some AI coding assistants——但Flask天生是同步的,写异步得靠各种补丁,性能上限就摆在那。

转折发生在去年双11。

那阵子公司搞活动,流量暴涨,我的Flask接口直接被打崩了。Gunicorn开了8个worker,CPU跑到90%+,接口响应时间从平时的200ms飙到了5秒开外。运维大哥半夜给我打电话:"哥,你那个服务是不是有内存泄漏?"

我排查了一宿,最后发现不是内存泄漏,是同步IO阻塞。一个数据库查询慢了,整个worker就卡住了。8个worker全卡住,服务就挂了。

那一刻我意识到,同步框架在高并发场景下就是原罪

为什么是FastAPI

其实我也看过Django、Tornado、aiohttp这些。Django太重了,我要的只是一个轻量API框架;Tornado文档劝退;aiohttp太底层,写起来费劲。

后来是组里一个大佬推荐了FastAPI。他当时说了一句话我印象特别深:"你写前端用Vue3的Composition API觉得爽吧?FastAPI写后端的体验差不多,而且自带Swagger文档,类型提示拉到满。"

我一开始是抵触的。

说实话,那阵子AI写代码的工具刚火起来,什么Copilot、文心一言的代码能力,我内心是有点排斥的。总觉得"代码得自己一行行敲才有灵魂"。但后来用FastAPI的时候,发现它的类型提示和自动补全体验太好了,好到让我重新思考"AI辅助写代码"这件事——这不是替代你,是让你少写那些无聊的样板代码。

扯远了,回到FastAPI。

选它的原因很简单:

维度 Flask FastAPI
异步支持 需额外配置 原生async/await
性能 一般 接近Node.js/Go
类型校验 需手动或用marshmallow 内置Pydantic
API文档 需flask-swagger等 自动生成Swagger/ReDoc
学习曲线 低(会Flask就会)

最打动我的是原生异步自动文档这两点。前者解决性能问题,后者解决我这种"最讨厌写文档"的懒人的痛点。

上手FastAPI:比想象中简单

安装就不说了,pip install fastapi uvicorn,两行命令的事。

直接看个例子,感受一下:

from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import Optional

app = FastAPI(title="我的破后端", version="0.1.0")

class ItemRequest(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

@app.post("/items/")
async def create_item(item: ItemRequest):
    # 这里item已经被Pydantic自动校验过了
    # 类型不对?直接返回422,都不用你写校验逻辑
    return {"message": "创建成功", "item": item}

@app.get("/items/{item_id}")
async def read_item(
    item_id: int,
    q: Optional[str] = Query(None, max_length=50)
):
    return {"item_id": item_id, "q": q}

看到没?没有一行代码在手动校验参数类型,没有一行代码在写Swagger文档。ItemRequest这个Pydantic模型,既是数据校验,又是API文档的数据结构说明。

启动服务:

uvicorn main:app --reload --host 0.0.0.0 --port 8000

然后打开 http://localhost:8000/docs,你会看到一个完整的Swagger UI,可以直接在页面上测试接口。

我当时看到那个页面的时候,内心OS是:以前那些手写Swagger配置的日子,算是白过了。

性能优化:这才是重头戏

光入门没用,咱得聊聊性能。毕竟我是被线上事故"打"过来的。

异步IO的正确打开方式

FastAPI的核心优势就是异步。但很多人(包括一开始的我)有个误区:加了async就完事了。

错。

如果你的异步函数里调用了同步的阻塞操作(比如requests.gettime.sleep、同步的数据库查询),那异步就是个摆设,甚至更慢——因为事件循环被阻塞了。

正确的做法:

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# ❌ 错误示范:在async函数里用同步requests
# import requests
# @app.get("/bad")
# async def bad_example():
#     resp = requests.get("https://api.example.com/data")  # 阻塞!
#     return resp.json()

# ✅ 正确示范:用异步httpx
@app.get("/good")
async def good_example():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://api.example.com/data")
        return resp.json()

# ✅ 如果非要用同步库,用run_in_executor
import requests
@app.get("/sync-but-ok")
async def sync_but_ok():
    loop = asyncio.get_event_loop()
    resp = await loop.run_in_executor(
        None, 
        lambda: requests.get("https://api.example.com/data")
    )
    return resp.json()

这个坑我踩了整整一周。当时接口加了async但性能没提升,我还以为FastAPI不行,后来用py-spy一profile,发现事件循环全被同步DB查询堵死了。

数据库连接池:别裸奔

用SQLAlchemy的话,一定要配连接池。我之前的Flask项目,每次请求都新建数据库连接,高并发下直接OOM。

FastAPI + SQLAlchemy的正确姿势:

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# 异步引擎,配合asyncpg驱动
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,          # 连接池大小
    max_overflow=10,       # 最大溢出连接
    pool_pre_ping=True,    # 自动检测失效连接
    pool_recycle=3600,     # 1小时回收连接
)

AsyncSessionLocal = sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

# 依赖注入,每个请求一个session
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

@app.get("/users/")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

这里有个细节:pool_pre_ping=True。这个参数会在每次从连接池取连接时,先发一个轻量级的ping检测连接是否还活着。生产环境必开,不然你下班后数据库那边把连接断了,第二天早上用户一访问,全报500。

别问我怎么知道的。

响应缓存:能缓存就缓存

有些接口数据变化不频繁,完全可以加缓存。FastAPI配合Redis做缓存非常简单:

import redis.asyncio as redis
import json
from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()
redis_client = redis.from_url("redis://localhost:6379", decode_responses=True)

@app.get("/hot-data")
async def get_hot_data():
    # 先查缓存
    cached = await redis_client.get("hot_data")
    if cached:
        return JSONResponse(json.loads(cached))
    
    # 缓存没有,查数据库
    data = await fetch_expensive_data()
    
    # 写入缓存,设置过期时间
    await redis_client.setex(
        "hot_data", 
        300,  # 5分钟过期
        json.dumps(data)
    )
    
    return data

加了缓存之后,那个之前P99 2800ms的接口,直接降到了50ms。Redis的读写速度是真的快。

中间件的坑

FastAPI的中间件是"洋葱模型",请求从外到内,响应从内到外。注意中间件的顺序很重要:

import time
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

# 注意顺序:先加的在外层
# CORS要在最外层
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# GZip压缩
app.add_middleware(GZipMiddleware, minimum_size=1000)

# 自定义耗时统计中间件
@app.middleware("http")
async def add_process_time_header(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

有个坑:自定义中间件里如果用了同步的time.sleep或者做了阻塞操作,整个服务都会卡住。中间件里一定要用异步操作。

生产环境部署:别在裸奔了

开发环境uvicorn --reload美滋滋,生产环境可不能这么搞。

我的部署方案是 Nginx + Uvicorn Workers

# 生产环境启动命令
uvicorn main:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 4 \
    --loop uvloop \
    --http httptools \
    --access-log

几个关键参数解释:

  • --workers 4:一般设置为CPU核心数的1-2倍。我们服务器是4核,就开4个worker
  • --loop uvloop:uvloop是基于libuv的事件循环,比默认的asyncio快很多,必装
  • --http httptools:更快的HTTP协议解析

Nginx配置:

upstream fastapi_backend {
    server 127.0.0.1:8000;
    # 如果有多个实例可以加更多server
}

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        
        # 超时配置
        proxy_connect_timeout 60s;
        proxy_read_timeout 120s;
    }
}

对了,别忘了用Supervisor或者systemd来管理uvicorn进程,保证挂了能自动重启。我之前有次服务器OOM,uvicorn直接死了,第二天用户投诉才发现。从那以后我就老老实实用systemd了。

一些进阶玩法

依赖注入系统

FastAPI的依赖注入是我最喜欢的特性之一,没有之一。它比Flask的g对象优雅太多了:

from fastapi import Depends, HTTPException, Header

# 认证依赖
async def get_current_user(
    authorization: str = Header(...)
):
    token = authorization.replace("Bearer ", "")
    user = await verify_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="认证失败")
    return user

# 权限依赖
async def require_admin(user: dict = Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(status_code=403, detail="权限不足")
    return user

# 数据库依赖
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# 组合使用
@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin),
    db: AsyncSession = Depends(get_db)
):
    await db.execute(delete(User).where(User.id == user_id))
    await db.commit()
    return {"message": "删除成功"}

依赖可以嵌套、可以复用、可以测试时mock。写大型项目的时候,这个特性简直是救命的。

后台任务

有些操作不需要等返回结果,比如发邮件、写日志、触发异步计算。FastAPI内置了BackgroundTasks:

from fastapi import BackgroundTasks

async def send_email_notification(email: str, message: str):
    """模拟发送邮件的异步任务"""
    await asyncio.sleep(3)  # 假设发邮件要3秒
    print(f"邮件已发送到 {email}: {message}")

@app.post("/register/")
async def register(
    email: str, 
    background_tasks: BackgroundTasks
):
    # 注册逻辑...
    user = await create_user(email)
    
    # 发邮件放到后台,不阻塞响应
    background_tasks.add_task(
        send_email_notification, 
        email, 
        "欢迎注册!"
    )
    
    return {"user_id": user.id}  # 立即返回,不等邮件发完

不过注意,BackgroundTasks适合轻量级的后台任务。如果是重计算或者长时间运行的任务,还是得上Celery或者RQ这种专门的任务队列。

关于AI辅助写代码这件事

写到最后,想聊点题外话。

开头说了,我一开始对AI写代码是抵触的。觉得那是偷懒,觉得代码得自己写才有"掌控感"。

但用了FastAPI之后,我发现一个有趣的事:当框架本身的类型提示和自动补全足够好的时候,AI辅助工具的效果也会变好。

FastAPI的代码天然就是强类型的,Pydantic模型、类型注解、依赖注入——这些信息对AI来说就是"上下文"。我用文心一言的代码助手来生成Pydantic模型或者CRUD接口的时候,准确率比以前用Flask高了一大截。因为FastAPI的代码结构更规范,AI更容易理解你的意图。

后来我甚至搞了个本地知识库,把公司的业务文档、数据库schema、接口规范都喂进去,配合Agentic AI的工作流,让AI帮我生成接口骨架代码。效果嘛,不能说100%可用,但80%的代码不用自己手写了,我只需要关注业务逻辑和边界情况。

说到这,推荐几本我看过觉得不错的书和资料:

  • 《FastAPI Web开发》—— 入门首选,讲得很系统
  • 《Python异步编程实战》—— 理解asyncio的底层原理
  • 《Designing Data-Intensive Applications》—— 虽然不是FastAPI专属,但做后端必读
  • FastAPI官方文档 —— 没错,官方文档就是最好的教程,写得比很多书都好

这些资料我存在了我的知识库里,没事就翻翻。

写在最后

从Flask到FastAPI,表面上是换了个框架,实际上是我对后端开发认知的一次升级。

以前我觉得后端就是"写写CRUD,调调数据库"。后来被线上事故毒打了一顿才明白,后端的核心是处理并发、管理资源、保证稳定性。FastAPI的异步模型、依赖注入、类型系统,都在帮你更好地做这些事。

现在我的VSCode里,Python插件依然装了一堆,但打开的项目里,前端和后端代码终于能和谐共处了。通勤的一号线地铁上,我甚至能用手机看看FastAPI的文档(别笑,真的有人这么干)。

如果你还在用Flask/Django写同步接口,被性能问题搞得焦头烂额,真心建议试试FastAPI。学习成本不高,迁移也不痛苦,但带来的性能提升和开发体验提升,是实打实的。

好了,不说了,产品经理又改需求了。这次他说要加个WebSocket实时推送——还好FastAPI原生支持,不用换框架了。

真香。


如果这篇文章对你有帮助,点个赞呗。要是你在迁移FastAPI的过程中踩了什么奇葩坑,评论区聊聊,让我开心开心。

评论 0

最热最新
暂无评论
阳光_思想家Lv.1
0
影响力
0
文章
0
粉丝