导语你用 FastAPI 写好了一套 CRUD 服务少量用户时跑得顺畅。直到某天流量上来p99 延迟一路攀升数据库连接数撞到上限几个接口开始变慢甚至超时。这时候你会意识到——同步写法撑不住高并发了。声明本文基于个人使用体验非商业推广。摘要本文从接口延迟升高、吞吐受限的真实痛点出发依次落地 async/await 异步基石、SQLAlchemy 异步连接池、Redis 缓存、BackgroundTasks 后台任务与限流最后给出 Uvicorn 多 worker 水平扩展方案并附 3 段可直接运行的完整代码。一、高并发下性能瓶颈一个请求进来后典型的 CRUD 接口要做三件事解析请求、查询数据库、序列化返回。低负载时整条链路几毫秒就跑完高并发时瓶颈往往不是 CPU而是等待。进程大部分时间卡在 I/O 上等数据库返回、等下游 HTTP 调用、等磁盘。同步接口在等待期间会占住整个工作线程于是少数慢查询就能耗光线程池把后续所有请求排队堵死。下面这张表概括了朴素实现里时间到底花在哪阶段同步实现耗时占比主要等待资源请求解析低CPU数据库查询高数据库连接序列化返回中CPU外部 HTTP 调用高网络核心结论吞吐量的上限取决于你能重叠多少个 I/O 等待而不是单个请求跑得多快。这正是 async 要解决的问题。二、异步基石 async/awaitasync/await是 Python 的协程coroutine可在等待点暂停、结束后恢复的函数语法async def声明的函数不会立刻执行而是返回一个可暂停的对象await表示在这里让出控制权等结果回来再继续。当多个协程都在await同一个会阻塞的 I/O 时事件循环event loop负责在协程间调度的运行时机抽象会在它们之间切换实现单线程内的并发。FastAPI 构建在ASGI异步服务器网关接口WSGI 的异步继任者之上因此async def路由能真正并发处理请求而普通def路由会被丢进线程池避免阻塞事件循环。下面这段代码同时给出两种路由写法异步路由直接await同步路由用普通defFastAPI 自动把它调度到线程池。# main.py from fastapi import FastAPI app FastAPI() app.get(/ping) async def ping(): # 异步路由await 让出控制权事件循环可去处理其他请求 return {msg: pong} app.get(/sync-work) def sync_work(): # 普通 def 路由FastAPI 自动放入线程池不阻塞事件循环 total sum(i * i for i in range(10_000)) return {result: total}注意在async def路由里不要调用阻塞式同步函数如time.sleep、同步requests.get否则会卡住整个事件循环。遇到 CPU 密集或阻塞工作用def路由或asyncio.to_thread包一层。三、数据库连接池优化 SQLAlchemy 异步引擎访问数据库是最典型的 I/O 等待。直接用同步驱动会在查询期间占住连接高并发时连接被打满。连接池connection pool一组可复用、用完即归还而非关闭的数据库连接能把连接数稳定在可控区间。SQLAlchemy 在 1.4 版本已引入异步支持、2.0 正式稳定pip install sqlalchemy[asyncio]会自动带上greenlet用于安装异步驱动。我们用asyncpg驱动配合create_async_engine构建异步引擎并通过参数控制池大小。# database.py from collections.abc import AsyncGenerator from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.orm import DeclarativeBase DATABASE_URL postgresqlasyncpg://user:passlocalhost:5432/app class Base(DeclarativeBase): pass # pool_size 约等于 CPU 核数的 2-4 倍max_overflow 约为其 2 倍 engine create_async_engine( DATABASE_URL, pool_size10, max_overflow20, pool_pre_pingTrue, ) AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) async def get_session() - AsyncGenerator[AsyncSession, None]: # 依赖注入每个请求拿到独立会话用完自动关闭 async with AsyncSessionLocal() as session: yield session async def dispose_engine() - None: # 应用关闭时释放所有连接避免连接泄漏 await engine.dispose()参数作用经验取值pool_size常驻连接数CPU 核数 × 2~4max_overflow峰值额外连接≈ pool_size × 2pool_pre_ping借出前探活Truepool_timeout等待连接超时默认 30s几个关键约束AsyncSession 不能跨任务或协程共享在多事件循环如多进程场景下使用NullPool避免跨循环共享连接应用退出务必调用engine.dispose()。四、缓存策略 Redis数据库查询再快也比内存慢一个数量级。对变化不频繁、读取频繁的数据如配置、热点列表用Redis内存键值数据库常作为缓存层挡在数据库前面可以大幅削峰。缓存读取遵循先查缓存命中即返回未命中再查库并回写的模式。下面用redis.asyncio客户端演示带超时与序列化的缓存封装。# cache.py import json from redis.asyncio import Redis redis Redis(hostlocalhost, port6379, decode_responsesTrue) async def get_user_profile(user_id: int) - dict: key fprofile:{user_id} cached await redis.get(key) if cached is not None: return json.loads(cached) # 缓存命中直接返回 # 缓存未命中查库此处用占位数据演示 profile {user_id: user_id, name: demo} await redis.set(key, json.dumps(profile), ex300) # 回写5 分钟过期 return profile缓存要警惕缓存穿透查不存在的 key 直击数据库和缓存雪崩大量 key 同时过期。常用对策对空结果也缓存短暂时间、给过期时间加随机抖动、用布隆过滤器前置拦截。更多异步客户端用法可参考 FastAPI 官方异步数据库文档外部官方参考。五、后台任务与限流有些操作不必阻塞响应发邮件、写审计日志、生成报表。FastAPI 的BackgroundTasks响应返回后再执行的任务队列可以把这类工作推迟到响应发出之后让接口先返回成功。# tasks.py import asyncio from fastapi import BackgroundTasks, FastAPI app FastAPI() # 限流信号量最多 5 个后台任务并发超出则排队 _semaphore asyncio.Semaphore(5) async def heavy_job(payload: dict) - None: async with _semaphore: await asyncio.sleep(1) # 模拟耗时工作 # 实际场景写库、调用第三方、生成文件等 print(done, payload) app.post(/submit) async def submit(payload: dict, bg: BackgroundTasks): bg.add_task(heavy_job, payload) # 先返回任务后台跑 return {status: accepted}asyncio.Semaphore在这里充当并发限流阀即便瞬间涌入大量请求后台任务也只以固定并发执行保护下游数据库与第三方服务不被打垮。把Semaphore换成令牌桶或滑动窗口还能做更细的接口级限流。关于异步并发模型与事件循环调度可延伸阅读 FastAPI Uvicorn 单线程异步与多 worker 并发模型详解站内相关文章。六、部署与水平扩展单机一个进程只能用一个 CPU 核跑事件循环无法利用多核。解决方法是多进程Uvicorn 多 worker用--workers N启动多个 worker 进程每个绑定一个事件循环能让 N 个核同时处理请求。# 启动 4 个 worker数量约等于 CPU 核数 uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4部署时留意几点多 worker 之间是独立进程不要共享内存状态例如用进程内字典做限流会失效限流计数应放到 Redis连接池在每个进程内独立存在总连接数 pool_size × workers数据库侧要留足余量再往上就是用 Nginx 或网关做负载均衡把流量分到多台机器实现真正的水平扩展。总结高并发优化不是堆机器而是先消除阻塞等待用async/await重叠 I/O用连接池稳定数据库连接用 Redis 挡掉重复查询用 BackgroundTasks 把非关键工作移出主路径最后用多 worker 与负载均衡吃满多核。每一项都配有完整代码可直接落地建议按先测基线 → 逐层加优化 → 压测对比的节奏推进。© 2026 | 转载请注明出处参考资料FastAPI 官方文档异步 SQL 数据库https://fastapi.tiangolo.com/zh/async/#async-sql-databases