fastapi怎么利用缓存优化性能？

FastAPI 可以通过多种缓存策略显著提升性能，尤其是在高并发或计算密集型场景下。以下是详细的优化方案和实现方法：

一、HTTP 响应缓存

1. 使用 Cache-Control头

from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int, response: Response):
    response.headers["Cache-Control"] = "public, max-age=3600"  # 浏览器缓存1小时
    return {"item_id": item_id, "data": "..."}

2. 中间件自动缓存

from fastapi.middleware.http import HTTPMiddleware

class CacheMiddleware:
    async def __call__(self, request, call_next):
        response = await call_next(request)
        if request.method == "GET":
            response.headers["Cache-Control"] = "public, max-age=300"
        return response

app.add_middleware(CacheMiddleware)

二、内存缓存（适合高频读取）

1. 使用 cachetools库

from cachetools import TTLCache
from fastapi import FastAPI

app = FastAPI()
cache = TTLCache(maxsize=1000, ttl=300)  # 最多缓存1000条，5分钟过期

@app.get("/expensive-calculation/{num}")
async def expensive_calculation(num: int):
    if num in cache:
        return {"result": cache[num], "source": "cache"}

    result = num ** 2  # 模拟耗时计算
    cache[num] = result
    return {"result": result, "source": "calculated"}

2. FastAPI 依赖注入缓存

from fastapi import Depends

def get_cache():
    return TTLCache(maxsize=1000, ttl=300)

@app.get("/data/{key}")
async def get_data(key: str, cache: TTLCache = Depends(get_cache)):
    if key in cache:
        return cache[key]
    # ... 数据库查询等操作

三、分布式缓存（适合集群环境）

1. Redis 集成

from redis import asyncio as aioredis
from fastapi import FastAPI

app = FastAPI()
redis = aioredis.from_url("redis://localhost:6379")

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    cache_key = f"user_{user_id}"
    cached_data = await redis.get(cache_key)
    if cached_data:
        return {"data": cached_data.decode(), "source": "redis"}

    # 数据库查询
    user_data = await db.fetch_user(user_id)  
    await redis.setex(cache_key, 3600, user_data.json())  # 缓存1小时
    return {"data": user_data, "source": "database"}

2. 使用 fastapi-cache插件

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend

@app.on_event("startup")
async def startup():
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

@app.get("/posts/{id}")
@cache(expire=60)  # 自动缓存60秒
async def get_post(id: int):
    return await db.get_post(id)

四、数据库查询缓存

1. SQLAlchemy 缓存

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy_cache import RedisCache

cache = RedisCache(redis)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

@app.get("/products/{id}")
async def get_product(id: int, session: AsyncSession = Depends(async_session)):
    stmt = select(Product).where(Product.id == id).options(cache.cache())
    result = await session.execute(stmt)
    return result.scalar_one()

五、CDN 缓存静态文件

from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")
# 配合Nginx配置：
# location /static {
#   expires 365d;
#   add_header Cache-Control "public";
# }