如果你正在用 FastAPI 开发一个 Web 服务并且天真地认为只要用了这个“高性能”框架你的接口响应速度就一定能快起来那你可能已经踩进了一个巨大的性能陷阱。我见过太多项目开发者兴冲冲地引入 FastAPI享受着它自动生成的交互式文档和简洁的语法却在不知不觉中把服务写成了“同步阻塞”模式。结果就是当并发请求稍微一上来服务器 CPU 看似悠闲但响应时间却直线飙升吞吐量惨不忍睹。问题出在哪很可能就出在你没有正确使用async和await。FastAPI 的“高性能”标签很大程度上是建立在它对异步Asynchronous编程的原生支持之上的。它底层基于 Starlette并默认使用 ASGI 服务器如 Uvicorn 或 Hypercorn。这套组合拳的核心优势就是能高效处理大量并发 I/O 操作如数据库查询、调用外部 API、读写文件。但如果你在路径操作函数前漏掉了async关键字或者在一个async def函数内部使用了阻塞式调用那么恭喜你你亲手把 FastAPI 这辆跑车开成了拖拉机性能直接“入土”。这篇文章不会只告诉你“要加async”而是要彻底讲清楚为什么在 FastAPI 中async对性能有如此决定性的影响什么情况下必须用async什么情况下可以用普通的def用了async就万事大吉了吗有哪些隐藏的坑会让你“异步了个寂寞”如何诊断和优化你项目中潜在的异步性能问题我们将从原理到实践通过代码对比和性能测试让你真正掌握 FastAPI 的高性能之道避免让你的服务在关键时刻“躺平”。1. 核心问题同步与异步在 FastAPI 中的天壤之别要理解async的重要性首先要抛开“异步就是多线程”的旧观念。在 Python 的asyncio世界里我们谈论的是单线程事件循环。想象一个场景你的 FastAPI 接口需要调用一个外部支付网关这个网络请求可能需要 2 秒才能返回。同时有 100 个用户发起请求。同步模式def函数 当一个请求进入这个同步函数时整个工作线程Worker会被完全阻塞呆呆地等待那 2 秒的网络 I/O 完成。在这 2 秒内这个线程不能处理任何其他请求。如果你用的是 Uvicorn 默认的线程池那么并发能力就受限于工作线程的数量。线程切换本身也有开销。100 个请求会排起长队平均等待时间极长。异步模式async def函数 当一个请求进入这个异步函数执行到await某个 I/O 操作如httpx.AsyncClient().get(...)时事件循环会立即挂起这个函数转而去处理其他已经就绪的任务比如处理另一个请求的解析或计算逻辑。等那个 2 秒的网络 I/O 完成后事件循环再回来继续执行这个函数剩下的部分。在这个过程中没有线程被阻塞一个线程就能同时处理成千上万个这样的并发连接。FastAPI 的默认服务器 Uvicorn 通常运行在单进程、单线程一个主事件循环模式下。它的高性能正是依赖于所有耗时操作都是“可等待的”Awaitable这样事件循环才能高效调度。如果你混入了同步阻塞操作就等于在这个高效运转的流水线上扔了一块大石头整个流水线都会被堵住。所以那个判断很简单如果你的路径操作函数内部包含任何 I/O 操作数据库、网络请求、磁盘读写并且你有性能需求那么async def几乎是唯一正确的选择。2. 环境准备与概念澄清在开始代码实战前我们先明确环境和几个关键概念。环境要求Python 3.7强烈建议 3.8以获得更稳定的asyncioFastAPI 最新版本Uvicorn 或 Hypercorn 作为 ASGI 服务器一个用于测试的 HTTP 客户端如httpx或requests安装依赖pip install fastapi uvicorn httpx sqlalchemy databases[aiosqlite]我们使用httpx作为异步 HTTP 客户端databases作为异步数据库驱动示例。关键概念澄清async defvsdefin FastAPI:async def: 声明一个异步函数。FastAPI 会将其放入事件循环中执行。函数内部可以使用await。def: 声明一个同步函数。FastAPI 会将其放入一个外部线程池中运行默认使用anyio.to_thread.run_sync以避免阻塞事件循环。但这引入了线程切换开销并非最佳性能路径。阻塞Blocking操作指会长时间占用 CPU 或等待 I/O 而不释放控制权的操作。例如time.sleep(5)同步休眠requests.get(...)同步 HTTP 请求open(...).read()同步文件读取某些同步数据库驱动如psycopg2的默认模式、mysql-connector-pythonCPU 密集型计算如复杂的数学运算、图像处理非阻塞/可等待操作指能够通知事件循环“我还没准备好你先去忙别的”的操作。例如asyncio.sleep(5)异步休眠httpx.AsyncClient().get(...)异步 HTTP 请求aiofiles.open(...).read()异步文件读取异步数据库驱动如asyncpg,aiomysql,databases库使用run_in_executor将 CPU 密集型任务丢到线程池。核心原则在async def函数内绝不能直接调用阻塞操作否则会阻塞整个事件循环让异步的优势荡然无存。3. 性能对比实验同步 vs 异步让我们用一个最经典的场景——调用外部慢 API——来直观感受差异。我们创建两个 FastAPI 端点/sync-api: 使用同步requests库模拟慢 I/O。/async-api: 使用异步httpx库模拟慢 I/O。项目结构fastapi-async-demo/ ├── main.py └── test_client.pymain.py- FastAPI 应用from fastapi import FastAPI import asyncio import time # 注意这里故意导入同步的 requests用于反面教材 import requests import httpx app FastAPI() # 模拟一个慢速的外部 API SLOW_API_URL https://httpbin.org/delay/2 # 这个地址会延迟2秒响应 app.get(/sync-ping) def sync_ping(): 完全同步的快速接口作为基线 return {message: pong, mode: sync} app.get(/sync-api) def call_slow_api_sync(): 【反面教材】在路径操作函数中使用同步阻塞调用。 这将阻塞工作线程。 start time.time() # 使用同步的 requests 库这是一个阻塞操作 response requests.get(SLOW_API_URL, timeout10) elapsed time.time() - start return {message: sync call done, status: response.status_code, time_elapsed: round(elapsed, 2)} app.get(/async-ping) async def async_ping(): 异步的快速接口 return {message: pong, mode: async} app.get(/async-api) async def call_slow_api_async(): 【正确示范】在异步函数中使用异步客户端。 在 await 期间事件循环可以处理其他任务。 start time.time() # 使用异步的 httpx 库 async with httpx.AsyncClient(timeout10.0) as client: response await client.get(SLOW_API_URL) elapsed time.time() - start return {message: async call done, status: response.status_code, time_elapsed: round(elapsed, 2)} app.get(/async-wrong) async def call_slow_api_async_but_wrong(): 【致命错误】在 async def 函数中使用了同步阻塞调用。 这会阻塞整个事件循环比同步端点更糟糕 start time.time() # 错误在异步函数中使用同步 requests response requests.get(SLOW_API_URL, timeout10) elapsed time.time() - start return {message: async def but sync call (WRONG!), status: response.status_code, time_elapsed: round(elapsed, 2)} if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)test_client.py- 并发测试客户端import asyncio import httpx import time import sys BASE_URL http://localhost:8000 CONCURRENCY 10 # 并发请求数 REQUEST_COUNT 20 # 总请求数 async def test_endpoint(endpoint: str, use_async_client: bool True): 测试单个端点的并发性能 print(f\n 测试端点: {endpoint} ) if use_async_client: # 使用异步客户端模拟高并发 async with httpx.AsyncClient(timeout30.0) as client: tasks [client.get(f{BASE_URL}{endpoint}) for _ in range(REQUEST_COUNT)] start time.time() responses await asyncio.gather(*tasks, return_exceptionsTrue) total_time time.time() - start success sum(1 for r in responses if not isinstance(r, Exception)) print(f异步客户端发起 {REQUEST_COUNT} 次请求 (并发度 ~{CONCURRENCY})) print(f总耗时: {total_time:.2f} 秒) print(f成功: {success}, 失败: {REQUEST_COUNT - success}) print(f平均每秒请求数 (RPS): {REQUEST_COUNT / total_time:.2f}) if success 0: # 计算平均响应时间 times [] for r in responses: if not isinstance(r, Exception) and r.status_code 200: try: data r.json() times.append(data.get(time_elapsed, 0)) except: pass if times: print(f接口平均响应时间: {sum(times)/len(times):.2f} 秒) else: # 使用同步客户端顺序执行作为对比 print(使用同步客户端顺序执行仅作参考) import requests start time.time() times [] for i in range(REQUEST_COUNT): try: r requests.get(f{BASE_URL}{endpoint}, timeout30) if r.status_code 200: data r.json() times.append(data.get(time_elapsed, 0)) except Exception as e: print(f请求 {i} 失败: {e}) total_time time.time() - start print(f顺序执行 {REQUEST_COUNT} 次请求) print(f总耗时: {total_time:.2f} 秒) print(f平均每秒请求数 (RPS): {REQUEST_COUNT / total_time:.2f}) if times: print(f接口平均响应时间: {sum(times)/len(times):.2f} 秒) async def main(): # 确保服务已启动 print(开始性能对比测试 (确保 FastAPI 服务运行在 localhost:8000)...) # 测试顺序快速接口 - 慢接口 await test_endpoint(/sync-ping, use_async_clientTrue) await test_endpoint(/async-ping, use_async_clientTrue) # 关键对比同步阻塞 vs 异步非阻塞 print(\n *50) print(核心性能对比处理慢 I/O 操作) print(*50) await test_endpoint(/sync-api, use_async_clientTrue) await test_endpoint(/async-api, use_async_clientTrue) # 警告这个端点会阻塞事件循环测试时谨慎并发数不要太高 print(\n !*50) print(警告以下测试将展示在 async def 中使用同步调用的灾难性后果) print(可能会暂时挂起服务测试后将恢复) print(!*50) try: # 只发少量请求以免服务完全卡死 global REQUEST_COUNT original_count REQUEST_COUNT REQUEST_COUNT 3 await test_endpoint(/async-wrong, use_async_clientTrue) REQUEST_COUNT original_count except Exception as e: print(f测试出错预期之中: {e}) print(\n测试完成。) if __name__ __main__: asyncio.run(main())如何运行与观察在一个终端启动服务uvicorn main:app --reload --host 0.0.0.0 --port 8000默认情况下Uvicorn 使用单进程单线程一个事件循环。在另一个终端运行测试python test_client.py预期结果分析你会看到类似下面的输出具体数值因机器和网络而异 测试端点: /sync-api 异步客户端发起 20 次请求 (并发度 ~10) 总耗时: 40.12 秒 成功: 20, 失败: 0 平均每秒请求数 (RPS): 0.50 接口平均响应时间: 2.01 秒 测试端点: /async-api 异步客户端发起 20 次请求 (并发度 ~10) 总耗时: 4.23 秒 成功: 20, 失败: 0 平均每秒请求数 (RPS): 4.73 接口平均响应时间: 2.02 秒/sync-api总耗时约 40 秒。因为同步视图函数call_slow_api_sync阻塞了工作线程。Uvicorn 默认的线程池可能只有少数几个工作线程处理这些阻塞调用请求需要排队等待可用线程导致总时间 ≈ 请求数 * 单次耗时 / 并发线程数。/async-api总耗时仅约 4 秒所有 20 个请求几乎同时发起每个请求在await client.get()时挂起事件循环同时管理所有网络连接。总时间 ≈ 单次耗时 少量开销。RPS每秒请求数是同步版本的近 10 倍/async-wrong这个端点可能会让测试客户端超时甚至导致服务暂时无响应。因为它用async def包装了同步阻塞调用直接卡死了唯一的事件循环线程。这个实验清晰地展示了对于 I/O 密集型操作正确的异步实现能将吞吐量提升一个数量级。4. 数据库操作同步驱动是性能杀手数据库访问是 Web 服务中最常见的 I/O 操作。错误地使用同步驱动如psycopg2的默认模式是 FastAPI 项目中最常见的性能问题之一。错误做法性能“入土”# 错误示例在 async def 中使用同步 ORM 驱动 from fastapi import FastAPI from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os DATABASE_URL postgresql://user:passlocalhost/dbname # 同步引擎和会话 engine create_engine(DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) name Column(String) app FastAPI() app.get(/users/{user_id}) async def get_user(user_id: int): # 警告在异步函数中使用同步 Session会阻塞事件循环 db SessionLocal() user db.query(User).filter(User.id user_id).first() db.close() return user上面的代码中db.query(...).first()是一个同步阻塞调用。当这个查询执行时事件循环会被卡住直到数据库返回结果。正确做法使用异步驱动这里以databasesasyncpgPostgreSQL为例。安装异步驱动pip install databases[postgresql] # 或 databases[mysql], databases[sqlite] # 对于 PostgreSQL还需要 asyncpg pip install asyncpg使用异步数据库会话from fastapi import FastAPI, Depends from sqlalchemy import Column, Integer, String, MetaData from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql import select import databases # 异步数据库连接 DATABASE_URL postgresqlasyncpg://user:passlocalhost/dbname database databases.Database(DATABASE_URL) # 定义元数据和模型使用 SQLAlchemy Core 或 ORM 的异步变体 metadata MetaData() # 方式1使用 SQLAlchemy Core推荐与 databases 库搭配 users Table( users, metadata, Column(id, Integer, primary_keyTrue, indexTrue), Column(name, String), ) # 方式2使用 ORM但查询时需使用 run_sync 或特定异步方法稍复杂 # 这里演示 Core 方式更直观 app FastAPI() app.on_event(startup) async def startup(): await database.connect() app.on_event(shutdown) async def shutdown(): await database.disconnect() app.get(/users/{user_id}) async def get_user(user_id: int): # 异步查询不会阻塞事件循环 query users.select().where(users.c.id user_id) user await database.fetch_one(query) return dict(user) if user else None # 更优雅的依赖注入方式 async def get_db(): # 通常不需要为每个请求创建新连接databases 有连接池 # 这里直接返回 database 对象 return database app.get(/users_v2/) async def list_users(db: databases.Database Depends(get_db), skip: int 0, limit: int 100): query users.select().offset(skip).limit(limit) results await db.fetch_all(query) return [dict(row) for row in results]关键点使用真正的异步驱动asyncpg,aiomysql,aiosqlite。使用支持异步的数据库库如databases,sqlalchemy.ext.asyncio。在所有数据库调用前使用await。利用连接池管理数据库连接。5. 处理 CPU 密集型任务不要阻塞事件循环异步擅长 I/O 密集型但不擅长 CPU 密集型。如果你在async def函数中执行大量计算如数据处理、图像渲染、复杂算法同样会阻塞事件循环。错误做法app.get(/heavy-computation) async def heavy_computation(): result 0 for i in range(10**7): # 一个耗时的 CPU 计算 result i * i return {result: result}这个循环会独占 CPU事件循环在此期间无法处理任何其他请求。正确做法使用asyncio.to_thread或run_in_executor将 CPU 密集型任务转移到单独的线程池中执行释放事件循环。import asyncio from concurrent.futures import ThreadPoolExecutor import time # 创建一个线程池执行器 executor ThreadPoolExecutor(max_workers4) # 根据 CPU 核心数调整 def cpu_bound_task(n: int) - int: 模拟一个 CPU 密集型任务 print(fCPU 任务开始参数: {n}) time.sleep(2) # 模拟计算耗时 result sum(i * i for i in range(n)) print(fCPU 任务结束结果: {result}) return result app.get(/compute/{n}) async def compute(n: int 1000000): 将 CPU 密集型任务卸载到线程池。 事件循环在任务运行期间可以处理其他 I/O。 # 方法1使用 asyncio.to_thread (Python 3.9) # result await asyncio.to_thread(cpu_bound_task, n) # 方法2使用 loop.run_in_executor (更通用) loop asyncio.get_event_loop() result await loop.run_in_executor(executor, cpu_bound_task, n) return {input: n, result: result} app.get(/compute-many) async def compute_many(): 并发执行多个 CPU 密集型任务。 注意线程池大小有限过多任务会排队。 tasks [] for i in range(1, 5): # 为每个任务创建异步任务 task asyncio.create_task( asyncio.get_event_loop().run_in_executor(executor, cpu_bound_task, i * 100000) ) tasks.append(task) results await asyncio.gather(*tasks) return {results: results}最佳实践建议区分任务类型I/O 密集型用async/awaitCPU 密集型用run_in_executor。控制线程池大小通常设置为 CPU 核心数或略多。考虑分离服务对于极重的计算任务考虑将其移出 Web 服务通过消息队列如 Celery Redis/RabbitMQ交给专门的 Worker 进程处理。6. 混合使用同步与异步代码安全边界现实项目往往不是纯异步的你可能需要调用一些尚未提供异步版本的第三方库。原则在异步上下文中调用同步代码必须将其“线程化”。import asyncio import requests from fastapi import FastAPI app FastAPI() def sync_http_call(url: str): 一个同步的第三方库调用假设没有异步版本 return requests.get(url).json() app.get(/data-from-sync-lib) async def get_data(): 安全地调用同步库。 # 将同步函数放到线程池中运行避免阻塞事件循环 loop asyncio.get_event_loop() data await loop.run_in_executor( None, # 使用默认的线程池执行器 sync_http_call, https://api.example.com/data ) return {data: data}警告频繁地将大量微小任务丢进线程池会因线程切换开销而抵消性能优势。如果可能优先寻找或封装异步版本的库。7. 常见问题与性能陷阱排查清单即使你知道了async/await的用法下面这些陷阱依然可能让你的 FastAPI 应用性能不佳。问题现象可能原因排查方式解决方案高并发下响应时间剧增但 CPU 使用率不高1. 路径操作函数是def而非async def且内部有 I/O。2. 在async def函数中使用了同步阻塞库如requests,psycopg2。1. 检查视图函数定义。2. 使用import语句检查使用的库是否为异步版本。3. 使用uvicorn的--log-level debug观察请求处理时间。1. 将def改为async def。2. 将同步库替换为异步库httpx,asyncpg,aiomysql。3. 或将同步调用用run_in_executor包装。服务完全无响应甚至健康检查都失败在async def函数中执行了长时间 CPU 计算或同步休眠time.sleep。检查代码中是否有循环计算、图像处理、time.sleep等。将 CPU 密集型任务移至线程池run_in_executor将time.sleep替换为asyncio.sleep。数据库查询慢连接数暴涨1. 使用了同步 ORM 驱动。2. 异步驱动但未使用连接池或连接池配置过小。1. 检查数据库驱动。2. 检查数据库连接字符串和池配置。1. 换用异步驱动。2. 正确配置连接池大小max_overflow,pool_size。async函数似乎没生效性能无改善1. 函数是async def但内部没有真正的await点全是 CPU 计算。2. 被await的操作本身是同步的如某些文件的open。1. 检查函数内部逻辑。2. 使用inspect.iscoroutinefunction验证。3. 使用性能分析工具如py-spy查看线程状态。1. 确保await的是真正的异步操作。2. 对文件 I/O 使用aiofiles。错误RuntimeError: Timeout context manager should be used inside a task在非异步上下文如普通def函数或__init__中使用了异步客户端或对象。检查错误堆栈找到初始化异步对象的位置。异步对象的初始化和生命周期管理应在异步上下文内如startup事件、依赖项。使用asyncio.run()或asyncio.create_task()来启动异步任务。日志中大量WARNING: ... Cancelling...客户端提前关闭了连接如超时但服务器端任务仍在运行。检查客户端超时设置和服务端任务是否过长。1. 合理设置客户端超时。2. 在服务端任务中添加可中断检查点如await asyncio.sleep(0)。3. 使用asyncio.shield保护关键任务谨慎使用。8. 最佳实践与工程化建议要让你的 FastAPI 应用真正发挥异步性能需要从架构到编码习惯上进行调整。从项目开始就确立异步优先原则在选择数据库驱动、HTTP 客户端、缓存客户端Redis、消息队列客户端时优先选择有良好维护的异步版本。常见选择httpx/aiohttp(HTTP),asyncpg/aiomysql(DB),redis[asyncio]/aioredis(Redis),aio-pika(RabbitMQ)。使用正确的 ASGI 服务器和配置生产环境使用uvicorn或hypercorn并配合多进程。# 使用多 worker 充分利用多核 CPU uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 # 或使用 Gunicorn 管理 Uvicorn worker更成熟 gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app调整连接数根据ulimit和数据库连接池限制调整--limit-concurrency和--backlog等参数。依赖注入与资源共享在startup事件中创建昂贵的异步客户端如 HTTP 客户端、数据库连接池并在shutdown中关闭。使用 FastAPI 的Depends来共享这些客户端避免为每个请求创建新连接。from fastapi import Depends, FastAPI import httpx app FastAPI() # 全局异步客户端需考虑线程安全httpx.AsyncClient 通常是安全的 async_client None app.on_event(startup) async def startup_event(): global async_client async_client httpx.AsyncClient(timeout10.0) app.on_event(shutdown) async def shutdown_event(): await async_client.aclose() async def get_http_client(): yield async_client app.get(/call-external) async def call_external(client: httpx.AsyncClient Depends(get_http_client)): r await client.get(https://api.example.com/data) return r.json()监控与 profiling使用uvicorn的访问日志和结构化日志如structlog,loguru来追踪请求生命周期。使用 APM 工具如OpenTelemetry,Sentry监控异步任务链。使用py-spy进行性能采样查看事件循环是否被阻塞。# 采样正在运行的 FastAPI 进程 py-spy top --pid PID_OF_UVICORN测试策略使用pytest和pytest-asyncio编写异步测试。使用asyncio的Mock或unittest.mock.AsyncMock来模拟异步依赖。进行压力测试时使用异步的负载测试工具如locust或async版本的wrk。9. 总结让 FastAPI 真正“快”起来的关键回到最初的问题为什么没加async性能就“入土”根本原因在于 FastAPI 的默认运行模式是基于单线程事件循环的 ASGI 服务器。它的高并发能力依赖于所有耗时操作都能“让出”控制权给事件循环。同步阻塞操作破坏了这一基石。要让你的 FastAPI 应用发挥极致性能请牢记以下行动清单对所有 I/O 密集型路径操作使用async def。在async def函数内部只使用异步库或通过run_in_executor包装同步调用。数据库操作务必使用异步驱动和连接池。CPU 密集型任务显式地转移到线程池。从项目依赖选择开始就建立异步技术栈。使用正确的服务器配置多 Worker来利用多核 CPU。通过监控和性能分析工具持续检查事件循环健康度。异步编程需要思维模式的转变从“顺序执行”到“协作式多任务”。一旦掌握你不仅能提升 FastAPI 应用的性能更能写出更高效、更可扩展的后端服务。现在就去检查你的项目里有没有那些“漏网”的同步阻塞调用吧。 新乡本地动态 ← 返回资讯列表
