在构建生产环境中的 FastAPI 应用时，性能是一个非常重要的考量因素。随着用户量的增加，应用的性能瓶颈会逐渐暴露出来。为了解决这些问题，可以通过异步操作、调优启动时间和响应时间、以及使用缓存来提高应用的性能。
1. 使用异步操作提高并发性能
FastAPI 本身是基于 ASGI（Asynchronous Server Gateway Interface）的，这意味着它可以很好地支持异步操作，并能够有效地处理高并发请求。通过使用异步操作，可以显著提高应用的并发能力，尤其是在处理 IO 密集型任务时。
1.1 异步视图函数
FastAPI 提供了原生支持异步视图函数。你可以使用 async 和 await 关键字来定义异步函数，这样在执行耗时的 IO 操作（如数据库查询、外部 API 调用等）时，FastAPI 会自动释放线程，允许其他请求处理。
示例：异步处理数据库查询
from fastapi import FastAPI
import asyncioapp = FastAPI()# 模拟一个耗时的异步操作
async def get_data_from_database():await asyncio.sleep(2) # 模拟延迟return {"data": "sample data"}@app.get("/data")
async def get_data():data = await get_data_from_database()return data
代码解析：
async def get_data_from_database()：定义了一个异步的数据库查询函数，使用await来模拟延迟。async def get_data()：处理请求的视图函数，它也是异步的，可以通过await调用异步数据库查询。
1.2 异步任务的优势
使用异步操作后，FastAPI 可以处理更多的并发请求，因为它不会被单个长时间运行的操作阻塞。异步 I/O 操作使得在等待外部资源时，应用可以继续处理其他请求，极大地提高了应用的并发性能。
- 异步数据库查询：如果你使用的是像
SQLAlchemy或Tortoise ORM这样的支持异步的数据库库，可以显著提高数据库操作的效率。 - 异步外部 API 调用：如果你的应用需要调用外部 API，异步调用可以使得应用在等待外部响应时，继续处理其他请求。
1.3 异步文件 I/O 操作
类似地，异步操作对于文件 I/O 也是有效的，尤其是在处理大型文件时。使用 aiofiles 库可以实现异步文件操作。
pip install aiofiles
python复制编辑import aiofiles@app.get("/read-file")
async def read_file():async with aiofiles.open("large_file.txt", mode='r') as file:content = await file.read()return {"file_content": content}
1.4 使用异步的 WebSocket 处理高并发
如果你需要支持 WebSocket，可以使用 FastAPI 的异步 WebSocket 端点，来处理实时双向通信。
from fastapi import WebSocket, FastAPIapp = FastAPI()@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data = await websocket.receive_text()await websocket.send_text(f"Message text was: {data}")
2. 调优 FastAPI 应用的启动时间和响应时间
2.1 启动时间优化
在 FastAPI 中，启动时间通常与应用初始化、依赖注入、数据库连接等因素相关。为了优化启动时间，可以考虑以下方法：
- 减少启动时的计算量：避免在启动时进行复杂的计算任务或加载大量数据。尽量将这些操作推迟到请求处理时进行。
- 延迟加载依赖：如果你的应用依赖于一些外部资源（如数据库、缓存等），可以考虑在首次请求时再加载这些资源，而不是在应用启动时就加载。
- 优化依赖注入：避免在应用启动时加载不必要的资源或进行不必要的依赖注入操作。
示例：延迟加载数据库连接
from fastapi import FastAPI, Dependsapp = FastAPI()# 模拟一个数据库连接类
class Database:def __init__(self):self.connection = Nonedef connect(self):self.connection = "Connected"def close(self):self.connection = None# 延迟加载数据库连接
def get_database():db = Database()db.connect() # 模拟连接yield dbdb.close() # 关闭连接@app.get("/items/")
async def read_items(db: Database = Depends(get_database)):return {"message": "Items fetched from DB", "connection": db.connection}
2.2 响应时间优化
响应时间优化直接影响用户体验和应用的吞吐量。以下是几个常见的优化方式：
- 优化数据库查询：对于数据库查询，尽量使用索引，避免全表扫描，减少不必要的查询。可以使用异步数据库库（如
databases和Tortoise ORM）来加速查询。 - 减少不必要的数据：避免在响应中返回大量数据，只返回客户端需要的数据。可以使用分页、过滤和投影来优化数据返回。
- 非阻塞 I/O：尽量避免使用同步 I/O 操作，尤其是长时间阻塞的操作，比如文件读写、网络请求等。可以使用异步 I/O 操作来解决这个问题。
- 批量操作：在可能的情况下，使用批量操作（例如批量数据库插入或更新），避免进行大量的单个操作。
3. 使用缓存提高性能
缓存是一种减少数据库查询和计算开销的常见技术。通过缓存，可以将频繁请求的数据存储在内存中，避免重复的计算或数据库查询。
3.1 使用 Redis 缓存
Redis 是一种高效的内存键值存储系统，广泛用于缓存场景。你可以使用 aioredis 库与 Redis 配合使用异步缓存。
安装 Redis 和 aioredis：
pip install aioredis
示例：使用 Redis 缓存数据
import aioredis
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()# Redis 客户端配置
redis = aioredis.from_url("redis://localhost", encoding="utf-8", decode_responses=True)class Item(BaseModel):name: strdescription: str@app.get("/items/{item_id}")
async def read_item(item_id: int):# 尝试从缓存中获取数据cached_item = await redis.get(f"item:{item_id}")if cached_item:return {"item": cached_item, "source": "cache"}# 如果缓存中没有数据，查询数据库（模拟）item = {"item_id": item_id, "name": "Item", "description": "Description"}# 将查询到的数据缓存到 Redisawait redis.set(f"item:{item_id}", str(item), ex=60) # 缓存过期时间 60 秒return {"item": item, "source": "database"}
代码解析：
- 使用
aioredis与 Redis 连接。 - 尝试从 Redis 中获取缓存的数据。如果缓存存在，就直接返回缓存数据。
- 如果缓存中没有数据，从数据库中查询并返回，并将数据存入缓存。
3.2 缓存响应数据
除了缓存单个数据，还可以考虑缓存整个响应（尤其是 GET 请求），例如使用 FastAPI-Cache 库来缓存响应。
pip install fastapi-cache
通过装饰器缓存 FastAPI 视图函数的响应：
from fastapi_cache.decorator import cacheable
from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
@cacheable(expire=60) # 缓存 60 秒
async def read_item(item_id: int):return {"item_id": item_id, "name": "Item", "description": "Description"}
通过以下几种方式，你可以显著优化 FastAPI 应用的性能：
- 使用异步操作：通过
async和await关键字优化数据库查询、文件操作和外部 API 调用等 I/O 密集型任务，提高并发性能。 - 调优启动时间和响应时间：减少启动时的计算、优化依赖注入、精简数据库查询和数据返回，提高响应时间。
- 使用缓存：通过缓存频繁请求的数据，减少数据库访问，提升响应速度。可以使用 Redis 或其他缓存技术来实现高效缓存。
这些优化措施可以帮助你构建高性能、高并发的 FastAPI 应用，提供更好的用户体验和更高的系统吞吐量。
