大家好，FastAPI已成为构建Python API的最流行框架之一，因其速度和易用性而广受欢迎。但在构建高性能应用程序时，使用同步（sync）还是异步（async）代码执行是很重要的问题。本文将通过现实世界的性能测试，对同步和异步的FastAPI实现进行基准测试，并深入分析相关数据，以帮助大家决定何时使用这两种方法。

1.同步与异步在FastAPI中的区别

• 同步代码（sync）：在同步执行中，任务一个接一个地处理。每个请求都需要等待前一个请求完成，这在用户数量较多或存在慢I/O操作（如数据库查询或文件上传）时，可能会导致瓶颈。

• 异步代码（async）：异步执行支持多个请求并发处理。应用程序无需等待I/O操作（如数据库调用）完成，而是可以继续处理其他请求，从而在高并发环境中更有效率。

2.设置：在FastAPI中基准测试同步与异步

为了比较同步和异步实现，在这里创建了两个版本的FastAPI应用程序。

同步版本：使用传统的阻塞数据库查询，采用psycopg2。

异步版本：使用非阻塞的异步查询，采用asyncpg。

这两个版本都执行一个简单的任务：从PostgreSQL数据库查询用户。数据库中包含极少的数据，以便隔离同步/异步机制的影响。

技术栈：

• 使用FastAPI作为API框架。

• 使用SQLAlchemy作为ORM。

• 使用psycopg2或psycopg2-binary（同步）和asyncpg（异步）进行数据库连接。

• 使用PostgreSQL作为数据库。

为了测试性能，使用**Apache Benchmark(ab)**模拟了1000个请求，具有100个并发连接。

2.1 同步版本代码

在同步版本中，在这里使用psycopg2驱动与SQLAlchemy，后者可执行阻塞查询。表格是使用同步的SQLAlchemy引擎创建的。

同步：main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import get_db, Userapp = FastAPI()@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):# 同步和阻塞user = db.query(User).filter(User.c.id == user_id).first()if not user:raise HTTPException(status_code=404, detail="User not found")return {"id": user.id, "name": user.name, "email": user.email}

同步：database.py

from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
from sqlalchemy.orm import sessionmakerDATABASE_URL = "postgresql://user:password@localhost/db_name"# 创建同步的SQLAlchemy引擎
engine = create_engine(DATABASE_URL, echo=True)# 创建用于同步查询的会话生成器
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)# 定义元数据
metadata = MetaData()# 定义User表
User = Table("users", metadata,Column("id", Integer, primary_key=True),Column("name", String),Column("email", String),
)# 在数据库中创建表
metadata.create_all(engine)# 获取同步数据库会话的依赖
def get_db():db = SessionLocal()try:yield dbfinally:db.close()

2.2 异步版本代码

在异步版本中，在这里使用asyncpg驱动与SQLAlchemy，进行非阻塞查询。然而，表的创建仍然是同步进行的，因为SQLAlchemy的metadata.create_all()不支持异步。

异步：main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from .database import get_async_db, User, initialize_database@asynccontextmanager
async def lifespan(app: FastAPI):# 启动：初始化数据库await initialize_database()yieldapp = FastAPI(lifespan=lifespan)@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):result = await db.execute(select(User).where(User.c.id == user_id))user = result.fetchone()if not user:raise HTTPException(status_code=404, detail="User not found")return {"id": user.id, "name": user.name, "email": user.email}

异步：database.py

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData, Table, Column, Integer, StringDATABASE_URL = "postgresql+asyncpg://user:password@localhost/db_name"# 用于异步查询的异步引擎
engine = create_async_engine(DATABASE_URL,echo=True,pool_size=10,max_overflow=20,
)# 用于异步查询的异步会话
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False
)# 定义元数据
metadata = MetaData()# 定义User表
users = Table("users",metadata,Column("id", Integer, primary_key=True),Column("name", String),Column("email", String),
)# 创建所有表
async def init_db():async with engine.begin() as conn:await conn.run_sync(metadata.create_all)# 获取异步数据库会话的依赖
async def get_async_db():async with AsyncSessionLocal() as session:yield session# 启动：初始化数据库
async def initialize_database():await init_db()

在这个版本中，请求是以异步方式处理的，支持多个请求在等待I/O时并发处理。

3.基准测试命令

同步版本：ab -n 1000 -c 100 http://127.0.0.1:8000/users/1

异步版本：ab -n 1000 -c 100 http://127.0.0.1:8001/users/1

以下是基准测试的性能指标分析：

同步vs异步（Airtable）的基准测试结果如下。

每秒请求数：异步版本每秒可处理50.68个请求，而同步版本每秒只能处理36.89个请求。在相同时间内，异步处理的请求数比同步版本多37%，因此在并发性方面，异步显然胜出。

每个请求的响应时间（平均值）：异步版本的平均响应时间低于同步版本27%（1973毫秒vs2710毫秒），这表明在高负载情况下，异步处理请求的效率更高。

最长请求时间：两个版本的最长请求时间相似（约4000毫秒），但异步版本的表现更稳定，这体现在响应时间的波动较小。

以上是同步和异步版本在不同百分位数下的比较图，包括平均和最长请求时间。

• 实线表示不同百分位数下的响应时间。

• 虚线表示同步（2710.648毫秒）和异步（1973.057毫秒）的平均响应时间。

• 点线突出显示同步（4167毫秒）和异步（3851毫秒）的最长请求时间。

4.FastAPI中同步与异步的选择

使用异步的情况：

• 应用程序需要处理高流量和大量并发用户。

• 应用程序是与I/O绑定的，需要进行大量数据库查询或API调用。

• 需要为大量请求最小化响应时间。

使用同步的情况：

• 应用程序的并发量较小，或主要执行CPU密集型任务。

• 希望保持代码库的简单性，避免异步处理的复杂性。

• 不希望应用程序扩展到同时处理数百或数千个请求。

5.优化异步性能

虽然异步在这些测试中速度更快，但仍有一些方法可以进一步优化。

• 连接池：使用连接池重用数据库连接，避免为每个请求创建一个新连接。

• 使用异步库：确保所有I/O绑定的任务（例如文件读/写、外部API调用）都以异步方式处理，以获得最佳性能。

• 测试更高的并发性：进行更高并发量的负载测试（例如，500+用户），以充分发挥异步的优势。

engine = create_async_engine(DATABASE_URL,pool_size=10,max_overflow=20
)

如果应用程序需要处理大量并发用户，并严重依赖于I/O绑定任务，那么异步FastAPI可以提供更好的性能、可扩展性和响应能力。不过对于更简单的用例，可以选择同步实现。