被裁后靠 FastAPI 接外包续命,一个老广程序员的性能优化血泪史
上周五晚上,十一点半,广州荔湾区西华路。
我坐在楼下那家开了二十年的牛杂档,要了一碗萝卜牛杂加两份面筋,老板阿叔看我天天来,已经不用我开口就知道我要"走青"(不要葱)。我一边吃一边盯着手机,微信上甲方发来的消息让我眉头紧锁:
"这个接口响应太慢了,3秒才返回,能不能优化一下?"
我叹了口气,把最后一块牛杂塞进嘴里。这已经是我这个月接的第三个外包项目了,用的都是 FastAPI。说实话,去年十月被裁的时候,我压根没想到自己有一天会靠 Python 后端框架来接活养家。
先自我介绍一下。我阿明,92年的老广,坐标广州老城区,做了快八年全栈开发。之前在天河软件园一家做 SaaS 的公司写代码,月薪从15k熬到22k,本以为能一直干到退休,结果公司"业务调整",整个后端组连我一起六个人,一锅端。
N+1 赔了三个月工资,到手八万多。听着不少对吧?但房贷每个月6800,房租(我住的地方租出去了自己搬去番禺)3500,老婆在幼儿园当老师月薪才5500,儿子明年要上小学,各种兴趣班一年得两三万。当时真的很焦虑,整夜整夜睡不着,头发掉得比写 Java 的时候还快。
一、为什么是 FastAPI?
被裁后头两个月,我疯狂投简历。35岁,在广州这个市场,你们懂的。面试了十几家,要么嫌我年龄大,要么嫌我要价高。有一家给了 offer,月薪16k,单休,大小周,我老婆说"不如不去"。
当时差点想放弃,甚至考虑过去开滴滴。但我一个做独立开发的朋友老陈跟我说:"你先接外包,用 FastAPI,现在好多小公司和个人项目都在用,上手快,性能好,异步天然支持。"
老陈是做 Go 的,他跟我说了一句让我印象很深的话:"你写 Python 的,如果不懂性能优化,那跟写脚本的有什么区别?FastAPI 的优势就是让你用 Python 的语法,拿到接近 Go 的性能体验。"
这话戳到我了。
我花了大概两周时间,把 FastAPI 的文档啃了一遍,又看了几十个 GitHub 上的开源项目。然后开始在程序员客栈、猪八戒、还有各种微信群里接单。
到今天,大半年过去了,我靠 FastAPI 接了七八个外包项目,月收入稳定在两万多,虽然比不上以前的22k,但时间自由,能接送儿子上下学,老婆也支持。
所以今天这篇文章,我想以一个"过来人"的身份,给 Python 后端新手聊聊 FastAPI,重点讲性能优化。毕竟,框架只是工具,写得好不好,看你优化到不到位。
二、FastAPI 入门:别被"快"字骗了
很多新手看到 FastAPI 的名字,以为用了就自动变快。兄弟,你想多了。
FastAPI 的"快",指的是框架本身的性能——基于 Starlette 和 Pydantic,底层用了 ASGI 协议,支持异步。但你的业务代码写得烂,该慢还是慢。
2.1 先跑起来
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
就这么简单。uvicorn main:app --reload 一跑,浏览器打开 http://127.0.0.1:8000,就能看到返回。
新手到这里就满足了,开始写业务逻辑。但我要告诉你,真正的坑从你写第一个数据库查询开始。
2.2 同步 vs 异步:这个必须搞清楚
FastAPI 支持 async def 和 def 两种路由函数。区别在哪?
# 异步路由
@app.get("/async")
async def async_route():
result = await some_async_function()
return result
# 同步路由
@app.get("/sync")
def sync_route():
result = some_sync_function()
return result
核心原则:如果你的路由函数里有 I/O 操作(数据库查询、HTTP 请求、文件读写),用 async def;如果是 CPU 密集型操作(图像处理、复杂计算),用 def,FastAPI 会自动把它丢到线程池里执行,不会阻塞事件循环。
我见过太多新手,在 async def 里直接调 requests.get() 或者用同步的 pymysql,把整个事件循环堵死。结果就是:一个请求慢,所有请求都慢。
记住这句话:异步函数里,只能用异步的库。
# 正确做法
import httpx
from sqlalchemy.ext.asyncio import AsyncSession
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
# 用 httpx 而不是 requests
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.example.com/data")
# 用 SQLAlchemy 的异步版本
result = await db.execute(select(User))
return result.scalars().all()
三、性能优化:这才是核心竞争力
接下来说重点。我接的外包项目,甲方最在意的就是性能。有个做电商的客户,他原来的 Django 项目,列表接口要 3 秒,换成 FastAPI 后我优化到 200ms,他直接多给我加了五千块。
3.1 数据库查询优化
问题一:N+1 查询
# 烂代码:N+1 问题
@app.get("/orders")
async def get_orders(db: AsyncSession = Depends(get_db)):
orders = await db.execute(select(Order))
result = []
for order in orders.scalars().all():
# 每次循环都查一次数据库!
user = await db.execute(select(User).where(User.id == order.user_id))
result.append({
"order": order,
"user": user.scalars().first()
})
return result
100 个订单,就要查 101 次数据库。不慢才怪。
优化方案:用 joinedload 或 selectinload
# 优化后:一次查询搞定
from sqlalchemy.orm import selectinload
@app.get("/orders")
async def get_orders(db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(Order)
.options(selectinload(Order.user))
)
return result.scalars().all()
一次查询,数据库压力直接降到原来的 1/100。
问题二:没有索引
这个太常见了。我接过一个项目,一张五十万数据的表,查询条件字段没加索引,每次查询全表扫描。加了个索引,响应时间从 2 秒降到 50ms。
# SQLAlchemy 模型里加索引
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String(100), index=True) # 加索引!
created_at = Column(DateTime, index=True) # 经常排序的字段也要加
3.2 缓存:能不动数据库就不动
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
# 初始化 Redis 缓存
redis = aioredis.from_url("redis://localhost")
FastAPICache.init(RedisBackend(redis), prefix="my-cache")
@app.get("/hot-products")
@cache(expire=60) # 缓存60秒
async def get_hot_products():
# 热门商品列表,不需要每次都查数据库
products = await db.execute(
select(Product).order_by(Product.sales.desc()).limit(20)
)
return products.scalars().all()
这里多说一句,我有个项目甲方要求用 Go 写微服务,但他们的知识库系统是用 Python 的。我就用 FastAPI 做 API 层,Redis 做缓存,Go 那边做计算密集型任务,两边通过 gRPC 通信。效果很好,各取所长。
3.3 响应压缩和序列化
FastAPI 默认用 Pydantic 做序列化,已经很快了。但如果你返回的数据量大,可以加上 gzip 压缩:
from fastapi import Response
from fastapi.middleware.gzip import GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
另外,Pydantic 模型里只返回需要的字段,别把整个数据库对象直接丢出去:
from pydantic import BaseModel
class UserResponse(BaseModel):
id: int
name: str
email: str
class Config:
from_attributes = True
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
user = await db.get(User, user_id)
return user # 只返回 UserResponse 里定义的字段
3.4 连接池:别每次都新建连接
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
# 配置连接池
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/dbname",
pool_size=20, # 连接池大小
max_overflow=10, # 最大溢出连接
pool_pre_ping=True, # 使用前检查连接是否有效
pool_recycle=3600, # 一小时回收连接
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
默认的连接池配置往往不够用,特别是并发高的时候。我一般会根据服务器配置和数据库承受能力来调整 pool_size。
四、实战踩坑:那些文档里没写的事
4.1 依赖注入的坑
FastAPI 的依赖注入很好用,但新手容易踩坑:
# 坑:每次请求都新建数据库 session,但忘了关闭
async def get_db():
db = AsyncSessionLocal()
yield db
# 忘了 await db.close()!
# 正确做法
async def get_db():
async with AsyncSessionLocal() as db:
yield db
4.2 后台任务的正确姿势
有些操作不需要等返回结果,比如发邮件、写日志:
from fastapi import BackgroundTasks
async def send_email(email: str, message: str):
# 模拟发送邮件
await asyncio.sleep(3)
print(f"Email sent to {email}")
@app.post("/register")
async def register(
user: UserCreate,
background_tasks: BackgroundTasks,
db: AsyncSession = Depends(get_db)
):
# 先存数据库
new_user = User(**user.dict())
db.add(new_user)
await db.commit()
# 邮件丢到后台任务
background_tasks.add_task(send_email, user.email, "Welcome!")
return {"message": "User created"}
但注意,BackgroundTasks 是在请求返回后执行的,如果服务重启,任务就丢了。重要的任务还是用 Celery 或者 RQ。
4.3 知识库项目的特殊优化
说到知识库,我上个月刚接了一个。甲方是个做教育的小公司,要做一个内部知识库系统,支持文档搜索、标签管理、权限控制。
技术栈:FastAPI + PostgreSQL + Elasticsearch + Redis
搜索这块我用了 Elasticsearch,全文检索比数据库 LIKE 快几十倍。但有个坑:ES 的数据同步。我用了 Canal 监听 MySQL binlog(后来改成 PostgreSQL 的 logical replication),实时同步到 ES。
# 搜索接口
@app.get("/knowledge/search")
async def search_knowledge(
q: str,
page: int = 1,
size: int = 20,
es: AsyncElasticsearch = Depends(get_es)
):
result = await es.search(
index="knowledge_base",
body={
"query": {
"multi_match": {
"query": q,
"fields": ["title^3", "content", "tags^2"]
}
},
"from": (page - 1) * size,
"size": size
}
)
return {
"total": result["hits"]["total"]["value"],
"items": [hit["_source"] for hit in result["hits"]["hits"]]
}
这个项目做完,甲方很满意,后来又给我介绍了两个客户。所以你看,技术做好了,口碑自然就来了。
五、FastAPI vs 其他框架:说点真话
我知道很多新手会纠结选什么框架。我简单说下我的看法:
| 框架 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| FastAPI | 快、类型安全、文档自动生成 | 生态不如 Django | API 服务、微服务 |
| Django | 全家桶、生态成熟 | 重、不够灵活 | 传统 Web 项目 |
| Flask | 轻量、灵活 | 需要自己搭积木 | 小项目、学习 |
| Go (Gin) | 性能最强、并发好 | 学习曲线陡 | 高并发、微服务 |
如果你做 Python 后端,FastAPI 目前是最推荐的选择。如果你追求极致性能,可以考虑 Go,但学习成本不低。我那个朋友老陈就是从 Python 转 Go 的,他说"回不去了",但我觉得语言只是工具,关键看场景。
六、写在最后
写这篇文章的时候,已经是凌晨一点了。儿子在隔壁房间睡着了,老婆今天值班不在家。我坐在番禺这间月租1800的单间里,听着窗外偶尔驶过的货车声,突然觉得生活虽然不容易,但还算有奔头。
被裁那年,我32岁。现在33了,头发又白了几根,但心态比那时候好多了。
给新手的几点建议:
别只学框架,要学原理。FastAPI 为什么快?因为 ASGI,因为 Pydantic,因为类型提示。理解了这些,你才能做好优化。
性能优化是系统工程。不是换个框架就万事大吉,数据库索引、缓存策略、代码写法,每个环节都要注意。
多接项目,多踩坑。文档看十遍不如实际写一遍。我现在的经验,全是一个个项目磨出来的。
保持学习,但不要焦虑。Go 火不火?火。但 Python 也不会死。把一门语言吃透,比什么都学一点强。
35岁不是终点。至少在广州,有技术、有经验、能扛事的程序员,永远有饭吃。
上个月,有个以前公司的同事找我,说他也被裁了,问我怎么接外包。我跟他说了句话:"别等被裁了才想退路,现在就开始积累。"
共勉。
如果你也是在做 Python 后端,或者也在接外包,欢迎加我微信交流。我是阿明,一个在广州老城区长大、在番禺租房写代码的中年程序员。
下一篇想写什么?评论区告诉我。如果反响好,我写一篇 FastAPI 部署到 Docker + Nginx 的实战教程。
—— 2024年6月15日,凌晨,于番禺


评论 0