被裁后靠 FastAPI 接外包续命,一个老广程序员的性能优化血泪史

监控面板盯梢人
2026-07-14 21:54
阅读 262

上周五晚上,十一点半,广州荔湾区西华路。

我坐在楼下那家开了二十年的牛杂档,要了一碗萝卜牛杂加两份面筋,老板阿叔看我天天来,已经不用我开口就知道我要"走青"(不要葱)。我一边吃一边盯着手机,微信上甲方发来的消息让我眉头紧锁:

"这个接口响应太慢了,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 defdef 两种路由函数。区别在哪?

# 异步路由
@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了,头发又白了几根,但心态比那时候好多了。

给新手的几点建议:

  1. 别只学框架,要学原理。FastAPI 为什么快?因为 ASGI,因为 Pydantic,因为类型提示。理解了这些,你才能做好优化。

  2. 性能优化是系统工程。不是换个框架就万事大吉,数据库索引、缓存策略、代码写法,每个环节都要注意。

  3. 多接项目,多踩坑。文档看十遍不如实际写一遍。我现在的经验,全是一个个项目磨出来的。

  4. 保持学习,但不要焦虑。Go 火不火?火。但 Python 也不会死。把一门语言吃透,比什么都学一点强。

  5. 35岁不是终点。至少在广州,有技术、有经验、能扛事的程序员,永远有饭吃。

上个月,有个以前公司的同事找我,说他也被裁了,问我怎么接外包。我跟他说了句话:"别等被裁了才想退路,现在就开始积累。"

共勉。


如果你也是在做 Python 后端,或者也在接外包,欢迎加我微信交流。我是阿明,一个在广州老城区长大、在番禺租房写代码的中年程序员。

下一篇想写什么?评论区告诉我。如果反响好,我写一篇 FastAPI 部署到 Docker + Nginx 的实战教程。

—— 2024年6月15日,凌晨,于番禺

评论 0

最热最新
暂无评论
监控面板盯梢人Lv.1
0
影响力
0
文章
0
粉丝