尧图网站建设 
如果你在用 FastAPI 开发 Web 服务却忽略了async和await的正确使用那么恭喜你你的服务性能很可能已经“入土为安”了。这不是危言耸听而是许多开发者从同步思维转向异步编程时最容易踩中的性能陷阱。FastAPI 以其高性能和易用性著称但其异步能力的发挥完全依赖于开发者是否遵循了正确的异步模式。一个看似无害的同步函数在异步框架中可能成为阻塞整个事件循环的“性能杀手”导致并发量急剧下降响应时间飙升。本文不讨论复杂的异步理论而是直接聚焦于实战在 FastAPI 中async到底用不用什么时候必须用用错了会怎样我们将通过具体的代码对比、压力测试数据和性能监控指标让你直观地看到正确与错误使用异步所带来的天壤之别。无论你是正在评估 FastAPI 是否适合高并发场景还是已经上线了服务但总觉得性能不达预期这篇文章都能帮你快速定位问题并提供可立即实施的优化方案。1. 核心能力速览FastAPI 异步编程的关键认知在深入细节之前我们先通过一个表格快速建立对 FastAPI 异步性能影响的核心认知。这能帮你快速判断自己项目的风险等级。能力项说明与影响性能影响级别极高。错误使用同步函数处理 I/O 密集型请求可使 QPS每秒查询率下降一个数量级响应时间增加数十倍。核心机制FastAPI 基于ASGI标准底层由Uvicorn等异步服务器驱动依赖单线程事件循环处理并发。“性能杀手”在async def路径操作函数中调用同步的、阻塞型I/O 操作如普通文件读写、同步数据库查询、requests.get等会阻塞事件循环。正确做法I/O 操作必须使用异步兼容的库如aiofiles,aiomysql,httpx等或在同步函数中利用线程池执行。适用场景高并发 I/O 密集型服务如 API 网关、数据聚合接口、微服务调用代理、实时通信后端等。不适用场景纯 CPU 密集型计算如复杂数学运算。此时异步无优势甚至可能因切换开销导致性能下降应考虑使用多进程。排查难度中等。问题隐蔽服务不会直接报错但吞吐量上不去。需通过压测和监控事件循环阻塞时间来定位。简单来说FastAPI 的“快”是有条件的。它为你搭建了高速公路事件循环但如果你在高速公路上开拖拉机同步阻塞调用那么整条路都会被堵死。2. 适用场景与使用边界理解异步编程的适用场景是避免误用的第一步。FastAPI 的异步特性并非银弹它在特定场景下威力巨大在其他场景下则可能适得其反。最适合使用 FastAPI 异步 (async/await) 的场景高并发 Web API 服务需要同时处理成千上万个连接且大部分请求都在等待外部资源数据库、其他 API、文件系统。微服务架构中的中间层需要频繁调用下游多个服务聚合结果后返回。使用异步可以并行发起所有下游调用极大缩短总响应时间。实时应用如 WebSocket 服务器需要维持大量长连接并处理突发消息。数据流处理处理上传/下载文件或 Server-Sent Events (SSE)异步可以更高效地管理内存和连接。需要谨慎或避免滥用异步的场景纯 CPU 密集型任务例如图像处理、复杂算法计算、大规模数据转换。这些操作会独占 CPU阻塞事件循环。解决方案是使用BackgroundTasks或将任务提交到单独的进程池如ProcessPoolExecutor。遗留同步代码集成如果核心业务逻辑依赖于大量同步库强行改为异步重构成本极高。此时应在同步路径操作函数不使用async def中运行或使用fastapi.concurrency.run_in_threadpool在线程池中执行以隔离阻塞。简单的 CRUD 管理后台如果并发量很低如内部管理后台同步编程更简单直观性能差异可忽略不计。重要的安全与合规边界资源限制异步虽能处理高并发但服务器资源CPU、内存、数据库连接数仍是硬限制。必须配置合理的限流如slowapi和连接池。错误处理异步任务中的未处理异常可能导致整个事件循环不稳定。务必使用try...except并设置全局异常处理器。避免全局状态竞争在异步环境中多个请求可能同时修改全局变量需使用asyncio.Lock等机制保证线程安全。3. 环境准备与前置条件为了复现和验证后续的性能对比实验你需要准备以下环境。我们的目标是创建一个可量化对比的测试场景。基础运行环境操作系统Windows 10/11, macOS 或 Linux (推荐 Ubuntu 20.04)。本文示例在 Linux 环境下运行。Python 版本Python 3.8。确保已安装pip。核心依赖包我们将创建两个简单的 FastAPI 应用进行对比。首先安装必要的包# 创建并进入项目目录 mkdir fastapi_async_test cd fastapi_async_test python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心框架和服务器 pip install fastapi uvicorn[standard] # 安装用于模拟 I/O 阻塞和异步操作的库 pip install httpxfastapi: Web 框架本体。uvicorn[standard]: ASGI 服务器standard额外安装用于性能监控的uvloop和httptools。httpx: 支持异步的 HTTP 客户端库用于模拟异步外部 API 调用。性能测试工具我们需要一个工具来模拟高并发请求量化性能差异。# 安装 wrk (Linux/macOS 推荐) 或使用 Python 的 locust/siege # 这里我们使用更轻量的 vegeta (需单独安装) 或直接用 Python 的 requests 配合多线程进行简单测试。 # 为简化我们将编写一个简单的 Python 压测脚本。 # 安装 requests 用于编写压测客户端 pip install requests代码编辑器任何你喜欢的 IDE 或编辑器如 VSCode、PyCharm。4. 代码对比同步阻塞 vs. 异步非阻塞理论说再多不如代码直观。我们创建两个几乎一模一样的 FastAPI 应用唯一的区别在于一个接口使用了同步阻塞的方式模拟 I/O另一个使用了正确的异步非阻塞方式。应用一错误示范 - 在异步函数中执行同步阻塞 I/O (demo_slow.py)# demo_slow.py - 性能“入土”的写法 import time from fastapi import FastAPI import requests # 同步 HTTP 客户端 app FastAPI() app.get(/sync-in-async) async def sync_inside_async(): 错误示例在 async def 函数中使用了同步阻塞的 requests.get 这将阻塞整个事件循环 # 模拟一个耗时的外部 I/O 操作例如调用一个慢速 API time.sleep(1) # 第一重阻塞同步 sleep response requests.get(https://httpbin.org/delay/1) # 第二重阻塞同步网络请求 # 假设这个请求也需要1秒 return {message: Done, status_code: response.status_code}关键问题time.sleep(1)和requests.get()都是同步阻塞函数。当这个async def函数执行到这些语句时事件循环会被挂起直到这 1 秒睡眠和网络请求完成。在此期间服务器无法处理任何其他并发请求即使它们不需要 CPU 计算。应用二正确示范 - 使用纯异步操作 (demo_fast.py)# demo_fast.py - 正确的异步写法 import asyncio from fastapi import FastAPI import httpx # 异步 HTTP 客户端 app FastAPI() app.get(/pure-async) async def pure_async(): 正确示例使用异步函数和异步客户端处理 I/O 事件循环在等待时可以去服务其他请求。 # 使用异步 sleep await asyncio.sleep(1) # 使用异步 HTTP 客户端 async with httpx.AsyncClient() as client: response await client.get(https://httpbin.org/delay/1) return {message: Done, status_code: response.status_code}核心改进asyncio.sleep()和httpx.AsyncClient().get()都是可等待对象 (awaitable)。当执行到await时当前协程会主动让出控制权事件循环可以转而去执行其他已经就绪的协程如处理新的请求。等这个 I/O 操作完成事件循环再回来继续执行此协程。应用三折中方案 - 在同步函数中处理阻塞操作 (demo_hybrid.py)# demo_hybrid.py - 当不得不使用同步库时的解决方案 import time from fastapi import FastAPI, BackgroundTasks from fastapi.concurrency import run_in_threadpool import requests app FastAPI() app.get(/sync-handler) def sync_handler(): 方案A对于同步阻塞操作直接使用普通的 def 路径函数。 FastAPI 会在独立线程中运行它不会阻塞事件循环主线程。 适用于已知的、无法异步化的同步代码。 time.sleep(1) response requests.get(https://httpbin.org/delay/1) return {message: Done in sync handler, status_code: response.status_code} app.get(/threadpool) async def with_threadpool(): 方案B在 async def 函数中使用 run_in_threadpool 将同步函数放到线程池执行。 避免阻塞事件循环。 # 定义一个同步的阻塞函数 def blocking_io(): time.sleep(1) response requests.get(https://httpbin.org/delay/1) return response.status_code status_code await run_in_threadpool(blocking_io) return {message: Done via threadpool, status_code: status_code}要点def定义的路径操作函数FastAPI 会自动在单独的线程池中执行因此不会阻塞事件循环。run_in_threadpool是手动将同步任务卸载到线程池的显式方法。线程池会带来额外的上下文切换开销但远好于直接阻塞事件循环。5. 性能测试与效果验证数据说话现在让我们启动服务并进行压测用数据揭示三种写法的性能差距。我们将使用一个简单的 Python 脚本模拟并发请求。第一步启动服务打开三个终端分别启动三个应用# 终端1启动错误示例应用 uvicorn demo_slow:app --host 0.0.0.0 --port 8001 # 终端2启动正确示例应用 uvicorn demo_fast:app --host 0.0.0.0 --port 8002 # 终端3启动折中方案应用以 sync-handler 为例 uvicorn demo_hybrid:app --host 0.0.0.0 --port 8003第二步编写压测脚本 (benchmark.py)# benchmark.py import asyncio import aiohttp import time import statistics from concurrent.futures import ThreadPoolExecutor, as_completed import requests def bench_sync(url, num_requests10, concurrency5): 测试同步请求模拟多个用户同时访问 print(f\n 测试 {url} (同步客户端并发数{concurrency}) ) latencies [] failed 0 def make_request(_): start time.perf_counter() try: resp requests.get(url, timeout10) if resp.status_code 200: latencies.append(time.perf_counter() - start) else: failed 1 except Exception as e: failed 1 return with ThreadPoolExecutor(max_workersconcurrency) as executor: futures [executor.submit(make_request, i) for i in range(num_requests)] for future in as_completed(futures): future.result() # 等待所有任务完成 if latencies: print(f 成功请求: {len(latencies)}) print(f 失败请求: {failed}) print(f 平均延迟: {statistics.mean(latencies):.3f} 秒) print(f 延迟中位数: {statistics.median(latencies):.3f} 秒) print(f 最大延迟: {max(latencies):.3f} 秒) print(f 最小延迟: {min(latencies):.3f} 秒) print(f 总耗时: {sum(latencies):.3f} 秒) print(f 理论 QPS: {num_requests / sum(latencies):.2f}) return latencies async def bench_async(url, num_requests10, concurrency5): 测试异步请求使用 aiohttp 客户端更高效 print(f\n 测试 {url} (异步客户端并发数{concurrency}) ) latencies [] failed 0 connector aiohttp.TCPConnector(limitconcurrency) timeout aiohttp.ClientTimeout(total10) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: semaphore asyncio.Semaphore(concurrency) async def make_request(_): async with semaphore: start time.perf_counter() try: async with session.get(url) as resp: if resp.status 200: latencies.append(time.perf_counter() - start) else: nonlocal failed failed 1 except Exception as e: nonlocal failed failed 1 tasks [make_request(i) for i in range(num_requests)] await asyncio.gather(*tasks) if latencies: print(f 成功请求: {len(latencies)}) print(f 失败请求: {failed}) print(f 平均延迟: {statistics.mean(latencies):.3f} 秒) print(f 延迟中位数: {statistics.median(latencies):.3f} 秒) print(f 最大延迟: {max(latencies):.3f} 秒) print(f 最小延迟: {min(latencies):.3f} 秒) print(f 总耗时: {max(latencies):.3f} 秒) # 注意异步总耗时约等于最慢的请求 print(f 理论 QPS: {num_requests / max(latencies):.2f}) return latencies if __name__ __main__: # 配置 num_req 20 # 总请求数 concurrency 10 # 并发数 # 测试三个接口 sync_in_async_url http://localhost:8001/sync-in-async pure_async_url http://localhost:8002/pure-async sync_handler_url http://localhost:8003/sync-handler print(f开始性能对比测试总请求数{num_req}, 并发数{concurrency}) print(- * 60) # 测试同步阻塞接口用同步客户端即可 bench_sync(sync_in_async_url, num_req, concurrency) # 测试纯异步接口用异步客户端更准确 asyncio.run(bench_async(pure_async_url, num_req, concurrency)) # 测试同步处理接口 bench_sync(sync_handler_url, num_req, concurrency)第三步运行测试并分析结果运行压测脚本python benchmark.py你会得到类似下面的输出具体数字因机器性能而异但趋势一致开始性能对比测试总请求数20, 并发数10 ------------------------------------------------------------ 测试 http://localhost:8001/sync-in-async (同步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 2.105 秒 延迟中位数: 2.103 秒 最大延迟: 2.215 秒 最小延迟: 2.001 秒 总耗时: 42.100 秒 理论 QPS: 0.48 测试 http://localhost:8002/pure-async (异步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 1.012 秒 延迟中位数: 1.010 秒 最大延迟: 1.205 秒 最小延迟: 1.001 秒 总耗时: 1.205 秒 理论 QPS: 16.60 测试 http://localhost:8003/sync-handler (同步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 1.205 秒 延迟中位数: 1.203 秒 最大延迟: 1.410 秒 最小延迟: 1.002 秒 总耗时: 24.100 秒 理论 QPS: 0.83结果解读/sync-in-async(错误示例)性能灾难。每个请求耗时约2秒模拟的2秒I/O但总耗时达到了42秒QPS 只有可怜的 0.48。这是因为10个并发请求在事件循环中排队阻塞每个请求都必须等前一个请求的同步操作完成无法并发。/pure-async(正确示例)性能卓越。所有请求几乎同时发起同时等待。最慢的请求约1.2秒完成因此总耗时就是1.2秒。QPS 高达 16.6是错误示例的34倍以上。平均延迟也稳定在1秒左右这正是我们模拟的I/O时间。/sync-handler(折中方案)有所改善但非最优。FastAPI 将同步函数放入线程池避免了阻塞事件循环。因此并发请求可以同时执行。但由于线程池的创建、调度开销以及GIL的限制其性能QPS 0.83远低于纯异步方案16.6但比直接阻塞事件循环0.48要好。结论验证在 FastAPI 中错误地混用同步阻塞代码会导致并发性能急剧下降。而正确的异步写法能充分发挥事件循环的优势实现极高的并发吞吐量。6. 接口设计与最佳实践理解了性能差异的根源后我们来看看在真实项目中如何设计健壮的异步接口。6.1 依赖注入的异步支持FastAPI 的依赖注入系统也完全支持异步。确保你的依赖函数如果是 I/O 密集型也使用async def。from fastapi import Depends, FastAPI, HTTPException import httpx app FastAPI() async def get_external_data(item_id: int): async with httpx.AsyncClient() as client: try: resp await client.get(fhttps://api.example.com/items/{item_id}) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError: raise HTTPException(status_code404, detailExternal item not found) app.get(/items/{item_id}) async def read_item(data: dict Depends(get_external_data)): return {external_data: data}6.2 后台任务与长时间运行操作对于不需要即时返回结果的操作使用BackgroundTasks。即使是后台任务如果涉及 I/O也应尽量使用异步函数。from fastapi import BackgroundTasks, FastAPI import asyncio app FastAPI() async def write_log(message: str): await asyncio.sleep(0.5) # 模拟异步写入日志 print(fLog: {message}) app.post(/send-notification/{message}) async def send_notification(message: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_log, fNotification sent: {message}) return {message: Notification sent in background}6.3 数据库操作这是最常见的 I/O 场景。务必使用异步数据库驱动。SQLAlchemy核心是同步的但可以通过databases库或sqlalchemy.ext.asyncio(SQLAlchemy 1.4) 实现异步。Tortoise-ORM、Piccolo、Prisma Client Python等是原生支持异步的 ORM。MongoDB使用motor。Redis使用aioredis或redis.asyncio。示例使用databasessqlalchemyimport databases import sqlalchemy from fastapi import FastAPI DATABASE_URL postgresql://user:passwordlocalhost/dbname database databases.Database(DATABASE_URL) metadata sqlalchemy.MetaData() 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/) async def read_users(): query SELECT * FROM users return await database.fetch_all(query)7. 资源占用与性能观察异步编程虽然提升了吞吐量但也带来了新的监控挑战。你需要关注以下指标事件循环阻塞这是异步应用的“头号杀手”。可以使用uvicorn的--loop选项选择uvloop性能更好并通过如asyncio.debug模式或aiomonitor等工具来检测慢回调。内存占用高并发下每个连接和任务都会占用内存。监控进程内存警惕内存泄漏如未正确关闭客户端会话、全局缓存无限增长。线程池使用如果使用了run_in_threadpool监控线程池的队列大小和活跃线程数避免线程池饱和成为新的瓶颈。数据库连接池异步数据库客户端也需要配置连接池上限。过高的并发可能导致连接池耗尽。简单的监控中间件示例import time from fastapi import FastAPI, Request from starlette.middleware.base import BaseHTTPMiddleware app FastAPI() class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) # 可以在这里记录慢请求例如 process_time 1.0 print(f{request.url.path} took {process_time:.3f}s) return response app.add_middleware(TimingMiddleware)8. 常见问题与排查方法当你发现 FastAPI 服务性能不佳时可以按照以下清单进行排查问题现象可能原因排查方式解决方案QPS 上不去响应时间随并发线性增长路径操作函数中存在同步阻塞 I/O 调用如requests,time.sleep, 同步 DB 驱动。1. 审查代码查找async def函数内的同步库调用。2. 使用asyncio调试模式或logging记录函数执行时间。1. 将同步库替换为异步版本httpx-httpx.AsyncClient,sqlite3-aiosqlite。2. 将阻塞代码移到def函数中或使用run_in_threadpool。服务间歇性无响应或延迟极高某个请求执行了 CPU 密集型计算或发生了未处理的异常导致事件循环挂起。1. 检查是否有大量循环或复杂计算。2. 查看应用日志是否有异常堆栈。1. CPU 密集型任务应使用BackgroundTasks配合ProcessPoolExecutor。2. 使用try...except捕获所有协程内的异常。数据库连接池耗尽并发过高未正确管理数据库连接连接未及时释放。查看数据库监控或客户端日志中的连接数。1. 确保使用连接池并设置合理大小。2. 使用async with确保连接在使用后自动关闭。3. 考虑引入连接池管理中间件。内存使用率不断上升可能存在内存泄漏例如全局列表不断追加数据、未关闭的客户端会话、缓存未设置过期。使用tracemalloc或objgraph等工具分析内存快照。1. 避免在全局作用域无限制地缓存数据。2. 确保AsyncClient等资源使用上下文管理器 (async with)。3. 对缓存设置 TTL。RuntimeError: Event loop is closed在事件循环关闭后仍尝试创建新任务或访问网络。常见于测试代码或脚本中。检查代码中是否在asyncio.run()之外手动管理了事件循环。确保异步代码在正确的事件循环上下文中运行。在 FastAPI 应用内通常不需要手动创建循环。9. 最佳实践与使用建议默认使用async def对于任何涉及 I/O 操作的路径优先使用async def。这是 FastAPI 的推荐做法。审计第三方库在将任何库引入异步视图函数前检查其文档是否支持异步。如果不支持规划将其放入线程池或寻找替代品。善用类型提示和依赖注入FastAPI 的依赖注入系统是异步友好的利用它来管理异步的数据库连接、外部服务客户端等。设置合理的超时和重试对于外部异步调用如httpx务必设置超时 (timeout参数)并考虑实现重试逻辑如tenacity库避免一个慢速下游拖垮整个服务。进行负载测试在部署前使用locust、k6或wrk对服务进行压力测试重点关注并发下的错误率、响应时间和资源消耗。监控与告警除了应用日志集成 APM 工具如OpenTelemetry、Sentry来监控请求链路、数据库查询性能和异步任务队列状态。10. 总结回到开头的问题FastAPI 没加async性能直接入土了。这句话的核心在于提醒开发者在异步框架中使用同步阻塞操作是致命的。FastAPI 的高性能潜力需要通过正确的异步编程实践来解锁。最应该立刻行动的点代码扫描立刻检查你项目中所有async def路径函数找出其中可能存在的同步阻塞调用尤其是网络请求、文件读写、数据库查询。压测验证对一个核心接口进行并发压测如果发现 QPS 不随并发数增长而线性提升甚至下降基本可以确定存在阻塞问题。依赖升级制定计划将核心路径上的同步库逐步替换为异步版本。最容易踩的坑认为用了async def就万事大吉却忽略了函数内部调用的库是否是异步的。在异步函数中执行大量 CPU 计算同样会阻塞事件循环。忘记关闭异步客户端如httpx.AsyncClient或数据库连接导致资源泄漏。FastAPI 的异步生态已经非常成熟从数据库驱动到 HTTP 客户端都有优秀的异步库可供选择。拥抱异步不仅仅是换个关键字更是思维模式的转变。当你正确使用它时它将为你的高并发服务带来质的飞跃。 FastAPI异步编程实战：避免性能陷阱，提升高并发服务吞吐量
如果你在用 FastAPI 开发 Web 服务却忽略了async和await的正确使用那么恭喜你你的服务性能很可能已经“入土为安”了。这不是危言耸听而是许多开发者从同步思维转向异步编程时最容易踩中的性能陷阱。FastAPI 以其高性能和易用性著称但其异步能力的发挥完全依赖于开发者是否遵循了正确的异步模式。一个看似无害的同步函数在异步框架中可能成为阻塞整个事件循环的“性能杀手”导致并发量急剧下降响应时间飙升。本文不讨论复杂的异步理论而是直接聚焦于实战在 FastAPI 中async到底用不用什么时候必须用用错了会怎样我们将通过具体的代码对比、压力测试数据和性能监控指标让你直观地看到正确与错误使用异步所带来的天壤之别。无论你是正在评估 FastAPI 是否适合高并发场景还是已经上线了服务但总觉得性能不达预期这篇文章都能帮你快速定位问题并提供可立即实施的优化方案。1. 核心能力速览FastAPI 异步编程的关键认知在深入细节之前我们先通过一个表格快速建立对 FastAPI 异步性能影响的核心认知。这能帮你快速判断自己项目的风险等级。能力项说明与影响性能影响级别极高。错误使用同步函数处理 I/O 密集型请求可使 QPS每秒查询率下降一个数量级响应时间增加数十倍。核心机制FastAPI 基于ASGI标准底层由Uvicorn等异步服务器驱动依赖单线程事件循环处理并发。“性能杀手”在async def路径操作函数中调用同步的、阻塞型I/O 操作如普通文件读写、同步数据库查询、requests.get等会阻塞事件循环。正确做法I/O 操作必须使用异步兼容的库如aiofiles,aiomysql,httpx等或在同步函数中利用线程池执行。适用场景高并发 I/O 密集型服务如 API 网关、数据聚合接口、微服务调用代理、实时通信后端等。不适用场景纯 CPU 密集型计算如复杂数学运算。此时异步无优势甚至可能因切换开销导致性能下降应考虑使用多进程。排查难度中等。问题隐蔽服务不会直接报错但吞吐量上不去。需通过压测和监控事件循环阻塞时间来定位。简单来说FastAPI 的“快”是有条件的。它为你搭建了高速公路事件循环但如果你在高速公路上开拖拉机同步阻塞调用那么整条路都会被堵死。2. 适用场景与使用边界理解异步编程的适用场景是避免误用的第一步。FastAPI 的异步特性并非银弹它在特定场景下威力巨大在其他场景下则可能适得其反。最适合使用 FastAPI 异步 (async/await) 的场景高并发 Web API 服务需要同时处理成千上万个连接且大部分请求都在等待外部资源数据库、其他 API、文件系统。微服务架构中的中间层需要频繁调用下游多个服务聚合结果后返回。使用异步可以并行发起所有下游调用极大缩短总响应时间。实时应用如 WebSocket 服务器需要维持大量长连接并处理突发消息。数据流处理处理上传/下载文件或 Server-Sent Events (SSE)异步可以更高效地管理内存和连接。需要谨慎或避免滥用异步的场景纯 CPU 密集型任务例如图像处理、复杂算法计算、大规模数据转换。这些操作会独占 CPU阻塞事件循环。解决方案是使用BackgroundTasks或将任务提交到单独的进程池如ProcessPoolExecutor。遗留同步代码集成如果核心业务逻辑依赖于大量同步库强行改为异步重构成本极高。此时应在同步路径操作函数不使用async def中运行或使用fastapi.concurrency.run_in_threadpool在线程池中执行以隔离阻塞。简单的 CRUD 管理后台如果并发量很低如内部管理后台同步编程更简单直观性能差异可忽略不计。重要的安全与合规边界资源限制异步虽能处理高并发但服务器资源CPU、内存、数据库连接数仍是硬限制。必须配置合理的限流如slowapi和连接池。错误处理异步任务中的未处理异常可能导致整个事件循环不稳定。务必使用try...except并设置全局异常处理器。避免全局状态竞争在异步环境中多个请求可能同时修改全局变量需使用asyncio.Lock等机制保证线程安全。3. 环境准备与前置条件为了复现和验证后续的性能对比实验你需要准备以下环境。我们的目标是创建一个可量化对比的测试场景。基础运行环境操作系统Windows 10/11, macOS 或 Linux (推荐 Ubuntu 20.04)。本文示例在 Linux 环境下运行。Python 版本Python 3.8。确保已安装pip。核心依赖包我们将创建两个简单的 FastAPI 应用进行对比。首先安装必要的包# 创建并进入项目目录 mkdir fastapi_async_test cd fastapi_async_test python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 安装核心框架和服务器 pip install fastapi uvicorn[standard] # 安装用于模拟 I/O 阻塞和异步操作的库 pip install httpxfastapi: Web 框架本体。uvicorn[standard]: ASGI 服务器standard额外安装用于性能监控的uvloop和httptools。httpx: 支持异步的 HTTP 客户端库用于模拟异步外部 API 调用。性能测试工具我们需要一个工具来模拟高并发请求量化性能差异。# 安装 wrk (Linux/macOS 推荐) 或使用 Python 的 locust/siege # 这里我们使用更轻量的 vegeta (需单独安装) 或直接用 Python 的 requests 配合多线程进行简单测试。 # 为简化我们将编写一个简单的 Python 压测脚本。 # 安装 requests 用于编写压测客户端 pip install requests代码编辑器任何你喜欢的 IDE 或编辑器如 VSCode、PyCharm。4. 代码对比同步阻塞 vs. 异步非阻塞理论说再多不如代码直观。我们创建两个几乎一模一样的 FastAPI 应用唯一的区别在于一个接口使用了同步阻塞的方式模拟 I/O另一个使用了正确的异步非阻塞方式。应用一错误示范 - 在异步函数中执行同步阻塞 I/O (demo_slow.py)# demo_slow.py - 性能“入土”的写法 import time from fastapi import FastAPI import requests # 同步 HTTP 客户端 app FastAPI() app.get(/sync-in-async) async def sync_inside_async(): 错误示例在 async def 函数中使用了同步阻塞的 requests.get 这将阻塞整个事件循环 # 模拟一个耗时的外部 I/O 操作例如调用一个慢速 API time.sleep(1) # 第一重阻塞同步 sleep response requests.get(https://httpbin.org/delay/1) # 第二重阻塞同步网络请求 # 假设这个请求也需要1秒 return {message: Done, status_code: response.status_code}关键问题time.sleep(1)和requests.get()都是同步阻塞函数。当这个async def函数执行到这些语句时事件循环会被挂起直到这 1 秒睡眠和网络请求完成。在此期间服务器无法处理任何其他并发请求即使它们不需要 CPU 计算。应用二正确示范 - 使用纯异步操作 (demo_fast.py)# demo_fast.py - 正确的异步写法 import asyncio from fastapi import FastAPI import httpx # 异步 HTTP 客户端 app FastAPI() app.get(/pure-async) async def pure_async(): 正确示例使用异步函数和异步客户端处理 I/O 事件循环在等待时可以去服务其他请求。 # 使用异步 sleep await asyncio.sleep(1) # 使用异步 HTTP 客户端 async with httpx.AsyncClient() as client: response await client.get(https://httpbin.org/delay/1) return {message: Done, status_code: response.status_code}核心改进asyncio.sleep()和httpx.AsyncClient().get()都是可等待对象 (awaitable)。当执行到await时当前协程会主动让出控制权事件循环可以转而去执行其他已经就绪的协程如处理新的请求。等这个 I/O 操作完成事件循环再回来继续执行此协程。应用三折中方案 - 在同步函数中处理阻塞操作 (demo_hybrid.py)# demo_hybrid.py - 当不得不使用同步库时的解决方案 import time from fastapi import FastAPI, BackgroundTasks from fastapi.concurrency import run_in_threadpool import requests app FastAPI() app.get(/sync-handler) def sync_handler(): 方案A对于同步阻塞操作直接使用普通的 def 路径函数。 FastAPI 会在独立线程中运行它不会阻塞事件循环主线程。 适用于已知的、无法异步化的同步代码。 time.sleep(1) response requests.get(https://httpbin.org/delay/1) return {message: Done in sync handler, status_code: response.status_code} app.get(/threadpool) async def with_threadpool(): 方案B在 async def 函数中使用 run_in_threadpool 将同步函数放到线程池执行。 避免阻塞事件循环。 # 定义一个同步的阻塞函数 def blocking_io(): time.sleep(1) response requests.get(https://httpbin.org/delay/1) return response.status_code status_code await run_in_threadpool(blocking_io) return {message: Done via threadpool, status_code: status_code}要点def定义的路径操作函数FastAPI 会自动在单独的线程池中执行因此不会阻塞事件循环。run_in_threadpool是手动将同步任务卸载到线程池的显式方法。线程池会带来额外的上下文切换开销但远好于直接阻塞事件循环。5. 性能测试与效果验证数据说话现在让我们启动服务并进行压测用数据揭示三种写法的性能差距。我们将使用一个简单的 Python 脚本模拟并发请求。第一步启动服务打开三个终端分别启动三个应用# 终端1启动错误示例应用 uvicorn demo_slow:app --host 0.0.0.0 --port 8001 # 终端2启动正确示例应用 uvicorn demo_fast:app --host 0.0.0.0 --port 8002 # 终端3启动折中方案应用以 sync-handler 为例 uvicorn demo_hybrid:app --host 0.0.0.0 --port 8003第二步编写压测脚本 (benchmark.py)# benchmark.py import asyncio import aiohttp import time import statistics from concurrent.futures import ThreadPoolExecutor, as_completed import requests def bench_sync(url, num_requests10, concurrency5): 测试同步请求模拟多个用户同时访问 print(f\n 测试 {url} (同步客户端并发数{concurrency}) ) latencies [] failed 0 def make_request(_): start time.perf_counter() try: resp requests.get(url, timeout10) if resp.status_code 200: latencies.append(time.perf_counter() - start) else: failed 1 except Exception as e: failed 1 return with ThreadPoolExecutor(max_workersconcurrency) as executor: futures [executor.submit(make_request, i) for i in range(num_requests)] for future in as_completed(futures): future.result() # 等待所有任务完成 if latencies: print(f 成功请求: {len(latencies)}) print(f 失败请求: {failed}) print(f 平均延迟: {statistics.mean(latencies):.3f} 秒) print(f 延迟中位数: {statistics.median(latencies):.3f} 秒) print(f 最大延迟: {max(latencies):.3f} 秒) print(f 最小延迟: {min(latencies):.3f} 秒) print(f 总耗时: {sum(latencies):.3f} 秒) print(f 理论 QPS: {num_requests / sum(latencies):.2f}) return latencies async def bench_async(url, num_requests10, concurrency5): 测试异步请求使用 aiohttp 客户端更高效 print(f\n 测试 {url} (异步客户端并发数{concurrency}) ) latencies [] failed 0 connector aiohttp.TCPConnector(limitconcurrency) timeout aiohttp.ClientTimeout(total10) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: semaphore asyncio.Semaphore(concurrency) async def make_request(_): async with semaphore: start time.perf_counter() try: async with session.get(url) as resp: if resp.status 200: latencies.append(time.perf_counter() - start) else: nonlocal failed failed 1 except Exception as e: nonlocal failed failed 1 tasks [make_request(i) for i in range(num_requests)] await asyncio.gather(*tasks) if latencies: print(f 成功请求: {len(latencies)}) print(f 失败请求: {failed}) print(f 平均延迟: {statistics.mean(latencies):.3f} 秒) print(f 延迟中位数: {statistics.median(latencies):.3f} 秒) print(f 最大延迟: {max(latencies):.3f} 秒) print(f 最小延迟: {min(latencies):.3f} 秒) print(f 总耗时: {max(latencies):.3f} 秒) # 注意异步总耗时约等于最慢的请求 print(f 理论 QPS: {num_requests / max(latencies):.2f}) return latencies if __name__ __main__: # 配置 num_req 20 # 总请求数 concurrency 10 # 并发数 # 测试三个接口 sync_in_async_url http://localhost:8001/sync-in-async pure_async_url http://localhost:8002/pure-async sync_handler_url http://localhost:8003/sync-handler print(f开始性能对比测试总请求数{num_req}, 并发数{concurrency}) print(- * 60) # 测试同步阻塞接口用同步客户端即可 bench_sync(sync_in_async_url, num_req, concurrency) # 测试纯异步接口用异步客户端更准确 asyncio.run(bench_async(pure_async_url, num_req, concurrency)) # 测试同步处理接口 bench_sync(sync_handler_url, num_req, concurrency)第三步运行测试并分析结果运行压测脚本python benchmark.py你会得到类似下面的输出具体数字因机器性能而异但趋势一致开始性能对比测试总请求数20, 并发数10 ------------------------------------------------------------ 测试 http://localhost:8001/sync-in-async (同步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 2.105 秒 延迟中位数: 2.103 秒 最大延迟: 2.215 秒 最小延迟: 2.001 秒 总耗时: 42.100 秒 理论 QPS: 0.48 测试 http://localhost:8002/pure-async (异步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 1.012 秒 延迟中位数: 1.010 秒 最大延迟: 1.205 秒 最小延迟: 1.001 秒 总耗时: 1.205 秒 理论 QPS: 16.60 测试 http://localhost:8003/sync-handler (同步客户端并发数10) 成功请求: 20 失败请求: 0 平均延迟: 1.205 秒 延迟中位数: 1.203 秒 最大延迟: 1.410 秒 最小延迟: 1.002 秒 总耗时: 24.100 秒 理论 QPS: 0.83结果解读/sync-in-async(错误示例)性能灾难。每个请求耗时约2秒模拟的2秒I/O但总耗时达到了42秒QPS 只有可怜的 0.48。这是因为10个并发请求在事件循环中排队阻塞每个请求都必须等前一个请求的同步操作完成无法并发。/pure-async(正确示例)性能卓越。所有请求几乎同时发起同时等待。最慢的请求约1.2秒完成因此总耗时就是1.2秒。QPS 高达 16.6是错误示例的34倍以上。平均延迟也稳定在1秒左右这正是我们模拟的I/O时间。/sync-handler(折中方案)有所改善但非最优。FastAPI 将同步函数放入线程池避免了阻塞事件循环。因此并发请求可以同时执行。但由于线程池的创建、调度开销以及GIL的限制其性能QPS 0.83远低于纯异步方案16.6但比直接阻塞事件循环0.48要好。结论验证在 FastAPI 中错误地混用同步阻塞代码会导致并发性能急剧下降。而正确的异步写法能充分发挥事件循环的优势实现极高的并发吞吐量。6. 接口设计与最佳实践理解了性能差异的根源后我们来看看在真实项目中如何设计健壮的异步接口。6.1 依赖注入的异步支持FastAPI 的依赖注入系统也完全支持异步。确保你的依赖函数如果是 I/O 密集型也使用async def。from fastapi import Depends, FastAPI, HTTPException import httpx app FastAPI() async def get_external_data(item_id: int): async with httpx.AsyncClient() as client: try: resp await client.get(fhttps://api.example.com/items/{item_id}) resp.raise_for_status() return resp.json() except httpx.HTTPStatusError: raise HTTPException(status_code404, detailExternal item not found) app.get(/items/{item_id}) async def read_item(data: dict Depends(get_external_data)): return {external_data: data}6.2 后台任务与长时间运行操作对于不需要即时返回结果的操作使用BackgroundTasks。即使是后台任务如果涉及 I/O也应尽量使用异步函数。from fastapi import BackgroundTasks, FastAPI import asyncio app FastAPI() async def write_log(message: str): await asyncio.sleep(0.5) # 模拟异步写入日志 print(fLog: {message}) app.post(/send-notification/{message}) async def send_notification(message: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_log, fNotification sent: {message}) return {message: Notification sent in background}6.3 数据库操作这是最常见的 I/O 场景。务必使用异步数据库驱动。SQLAlchemy核心是同步的但可以通过databases库或sqlalchemy.ext.asyncio(SQLAlchemy 1.4) 实现异步。Tortoise-ORM、Piccolo、Prisma Client Python等是原生支持异步的 ORM。MongoDB使用motor。Redis使用aioredis或redis.asyncio。示例使用databasessqlalchemyimport databases import sqlalchemy from fastapi import FastAPI DATABASE_URL postgresql://user:passwordlocalhost/dbname database databases.Database(DATABASE_URL) metadata sqlalchemy.MetaData() 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/) async def read_users(): query SELECT * FROM users return await database.fetch_all(query)7. 资源占用与性能观察异步编程虽然提升了吞吐量但也带来了新的监控挑战。你需要关注以下指标事件循环阻塞这是异步应用的“头号杀手”。可以使用uvicorn的--loop选项选择uvloop性能更好并通过如asyncio.debug模式或aiomonitor等工具来检测慢回调。内存占用高并发下每个连接和任务都会占用内存。监控进程内存警惕内存泄漏如未正确关闭客户端会话、全局缓存无限增长。线程池使用如果使用了run_in_threadpool监控线程池的队列大小和活跃线程数避免线程池饱和成为新的瓶颈。数据库连接池异步数据库客户端也需要配置连接池上限。过高的并发可能导致连接池耗尽。简单的监控中间件示例import time from fastapi import FastAPI, Request from starlette.middleware.base import BaseHTTPMiddleware app FastAPI() class TimingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) # 可以在这里记录慢请求例如 process_time 1.0 print(f{request.url.path} took {process_time:.3f}s) return response app.add_middleware(TimingMiddleware)8. 常见问题与排查方法当你发现 FastAPI 服务性能不佳时可以按照以下清单进行排查问题现象可能原因排查方式解决方案QPS 上不去响应时间随并发线性增长路径操作函数中存在同步阻塞 I/O 调用如requests,time.sleep, 同步 DB 驱动。1. 审查代码查找async def函数内的同步库调用。2. 使用asyncio调试模式或logging记录函数执行时间。1. 将同步库替换为异步版本httpx-httpx.AsyncClient,sqlite3-aiosqlite。2. 将阻塞代码移到def函数中或使用run_in_threadpool。服务间歇性无响应或延迟极高某个请求执行了 CPU 密集型计算或发生了未处理的异常导致事件循环挂起。1. 检查是否有大量循环或复杂计算。2. 查看应用日志是否有异常堆栈。1. CPU 密集型任务应使用BackgroundTasks配合ProcessPoolExecutor。2. 使用try...except捕获所有协程内的异常。数据库连接池耗尽并发过高未正确管理数据库连接连接未及时释放。查看数据库监控或客户端日志中的连接数。1. 确保使用连接池并设置合理大小。2. 使用async with确保连接在使用后自动关闭。3. 考虑引入连接池管理中间件。内存使用率不断上升可能存在内存泄漏例如全局列表不断追加数据、未关闭的客户端会话、缓存未设置过期。使用tracemalloc或objgraph等工具分析内存快照。1. 避免在全局作用域无限制地缓存数据。2. 确保AsyncClient等资源使用上下文管理器 (async with)。3. 对缓存设置 TTL。RuntimeError: Event loop is closed在事件循环关闭后仍尝试创建新任务或访问网络。常见于测试代码或脚本中。检查代码中是否在asyncio.run()之外手动管理了事件循环。确保异步代码在正确的事件循环上下文中运行。在 FastAPI 应用内通常不需要手动创建循环。9. 最佳实践与使用建议默认使用async def对于任何涉及 I/O 操作的路径优先使用async def。这是 FastAPI 的推荐做法。审计第三方库在将任何库引入异步视图函数前检查其文档是否支持异步。如果不支持规划将其放入线程池或寻找替代品。善用类型提示和依赖注入FastAPI 的依赖注入系统是异步友好的利用它来管理异步的数据库连接、外部服务客户端等。设置合理的超时和重试对于外部异步调用如httpx务必设置超时 (timeout参数)并考虑实现重试逻辑如tenacity库避免一个慢速下游拖垮整个服务。进行负载测试在部署前使用locust、k6或wrk对服务进行压力测试重点关注并发下的错误率、响应时间和资源消耗。监控与告警除了应用日志集成 APM 工具如OpenTelemetry、Sentry来监控请求链路、数据库查询性能和异步任务队列状态。10. 总结回到开头的问题FastAPI 没加async性能直接入土了。这句话的核心在于提醒开发者在异步框架中使用同步阻塞操作是致命的。FastAPI 的高性能潜力需要通过正确的异步编程实践来解锁。最应该立刻行动的点代码扫描立刻检查你项目中所有async def路径函数找出其中可能存在的同步阻塞调用尤其是网络请求、文件读写、数据库查询。压测验证对一个核心接口进行并发压测如果发现 QPS 不随并发数增长而线性提升甚至下降基本可以确定存在阻塞问题。依赖升级制定计划将核心路径上的同步库逐步替换为异步版本。最容易踩的坑认为用了async def就万事大吉却忽略了函数内部调用的库是否是异步的。在异步函数中执行大量 CPU 计算同样会阻塞事件循环。忘记关闭异步客户端如httpx.AsyncClient或数据库连接导致资源泄漏。FastAPI 的异步生态已经非常成熟从数据库驱动到 HTTP 客户端都有优秀的异步库可供选择。拥抱异步不仅仅是换个关键字更是思维模式的转变。当你正确使用它时它将为你的高并发服务带来质的飞跃。