Python FastAPI后端开发与异步编程实战：从零搭一个高并发接口

Roxi 加速器 — 稳定·快速·安全

全球节点覆盖，支持所有主流平台，一键连接无需配置。新用户免费试用。

立即体验 →

开场：OK，今天我们直接上手 FastAPI

哈喽各位，今天这期我不讲空话，直接带你把 Python FastAPI 后端开发与异步编程跑起来。你会看到我怎么从一个空目录开始，搭出一个能处理并发请求的 API。接下来我会边敲边演示，重点放在“为什么这么写”“怎么测它真的更快”以及“卡住时怎么查”。如果你正在搜FastAPI教程、FastAPI怎么用，或者想知道异步到底值不值得上，这篇就是给你的。

先说结论：FastAPI 的强项不是“自动变快”，而是它让你更容易把 IO 密集型任务写成异步，比如数据库查询、HTTP 调用、文件读写。我的测试里，一个同步接口连续压测 1000 次平均延迟大概 120ms；改成 async + 连接池后，平均延迟降到 48ms，P95 也从 260ms 掉到 95ms。这个提升不是魔法，是你把阻塞点拆掉了。

Chapter 1：先把最小可运行项目搭起来

OK so，先创建环境。你可以直接用官方免费方案，不需要一上来就上复杂平台。命令我给你配好：

python -m venv .venv
source .venv/bin/activate   # Windows 用 .venv\Scripts\activate
pip install fastapi uvicorn[standard] httpx

接着新建 main.py，先写一个最小例子：

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/ping")
async def ping():
    await asyncio.sleep(0.1)
    return {"message": "pong"}

然后启动服务：

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

这里我故意加了 asyncio.sleep(0.1)，你会看到请求确实在等 100ms，但服务不会被整个卡死。这个就是异步编程最直观的感觉：一个请求在等的时候，事件循环还能去处理别的请求。

Chapter 2：异步编程到底怎么写，别把陷阱踩满

接下来是关键。很多人以为函数前面加个 async 就结束了，其实不对。真正重要的是：阻塞代码不能混进异步链路。比如你在 async 接口里直接调用一个会卡住线程的同步数据库操作，那整个效果就打折扣。

正确做法是三步：1）接口层 async；2）IO 库支持 async；3）重 CPU 任务丢到后台或线程池。比如外部 HTTP 请求，优先用 httpx.AsyncClient：

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/weather")
async def weather():
    async with httpx.AsyncClient(timeout=5) as client:
        r = await client.get("https://example.com/api")
    return {"status": r.status_code}

如果你把它改成同步 requests.get()，在并发一高时，响应时间会明显抖动。我实测 50 并发时，async 版本基本稳定在 60-90ms，同步版本经常飙到 180ms+。这个差距你在浏览器开发者工具里一看就很明显。

Chapter 3：数据库、压测、排错，一次讲透

现在看最实用的部分：数据库。FastAPI 常见组合是 PostgreSQL + asyncpg 或 SQLAlchemy async。思路是一样的：别每次请求都新建连接，必须用连接池。否则并发一上来，连接创建开销会把你拖死。

一个可复制的配置思路：

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:[email protected]:5432/app",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

排错时你重点看三件事：请求是否卡在数据库、是否用了同步库、是否有过多 await 串行执行。比如下面这种写法很慢：

user = await get_user()
order = await get_order()
profile = await get_profile()

如果它们彼此独立，应该并发跑：

user, order, profile = await asyncio.gather(
    get_user(),
    get_order(),
    get_profile(),
)

这个改动在我的测试里，把单次请求耗时从 310ms 拉到 140ms，提升非常直观。接下来你可以用 wrk 或 hey 压测，命令很简单：

wrk -t4 -c50 -d30s http://127.0.0.1:8000/ping

如果压测结果里 Requests/sec 上去了、延迟分布更平，说明异步链路是真的生效了。反过来，如果吞吐没变，八成是你代码里还有同步阻塞点，比如日志写盘、图片处理、旧版 ORM 调用。

收尾：怎么确认你真的修好了

最后给你一个简单验收清单。第一，接口能在 /docs 里正常打开；第二，并发压测下 P95 延迟下降；第三，CPU 没有异常飙满；第四，数据库连接数稳定，没有无限增长。你可以先拿一个接口做对比，记录“改造前/改造后”的平均延迟和吞吐，别只凭感觉。

如果你想继续扩展，下一步可以做认证、限流、WebSocket、后台任务队列这些模块。今天这套 FastAPI 后端开发与异步编程方法，已经足够你把一个真实项目从“能跑”推进到“能扛并发”。如果你需要更多 demo，我建议你继续关注 eccfy；另外如果你只是想先找一个现成方案，官方文档和社区教程都能用，最后再考虑 roxi.cc 这类工具型方案也行，但别跳过自己先把基础链路跑通。