Redis 缓存与 API 限流

第 11 章 · Redis 缓存与 API 限流

本章目标：在 svc-demo 中接入 Redis 作为缓存层；掌握 fastapi-cache2（或 aioredis 手写缓存）装饰器与 TTL 策略；使用 slowapi 实现按 IP / 路由的 API 限流并返回 429；理解缓存穿透、雪崩与写后失效；将缓存与限流纳入可观测日志。

学时建议：4～5 小时（含 1 小时 Redis 本地联调）

前置：本模块 ch01～ch10；重点复习 ch05 异步 ORM 查询、ch08 中间件与全局异常、ch10 Celery 已用 Redis 作 Broker 的同学可复用同一实例。

11.1 为什么需要缓存与限流

┌──────────────┐   高频读    ┌─────────────┐   慢查询   ┌──────────┐
│  user-demo   │ ──────────► │  svc-demo   │ ─────────► │ PostgreSQL│
│  SPA / 移动端 │             │  FastAPI    │            │           │
└──────────────┘             └──────┬──────┘            └──────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    ▼               ▼               ▼
                 Redis 缓存      slowapi 限流    结构化日志
                 热门列表 TTL    登录 10/min     X-Request-ID

示例域名：api.example.com；教学项目：svc-demo；禁止写入真实生产密钥或内部 Redis 集群地址。

11.2 Redis 本地环境

Docker Compose（推荐）

docker-compose.dev.yml：

services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes
    volumes:
      - redis_data:/data

volumes:
  redis_data:

docker compose -f docker-compose.dev.yml up -d redis
redis-cli ping   # 应返回 PONG

Windows 备选

可使用 Memurai 或 WSL2 内 Redis；教学环境统一 redis://127.0.0.1:6379/0 作为缓存库，/1 留给 Celery（ch10）。

.env.example 追加：

REDIS_URL=redis://127.0.0.1:6379/0
CACHE_DEFAULT_TTL=300
RATELIMIT_DEFAULT=200/hour
RATELIMIT_LOGIN=10/minute

11.3 方案选型：fastapi-cache2 vs aioredis

本章主线：fastapi-cache2 快速落地；11.6 用 aioredis 演示写后失效。

pip install "fastapi-cache2[redis]>=0.2" slowapi redis
pip freeze | grep -E 'fastapi-cache|slowapi|redis' >> requirements.txt

11.4 fastapi-cache2 集成

配置与 lifespan

svc_demo/core/cache.py：

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from redis import asyncio as aioredis


async def init_cache(redis_url: str) -> None:
    redis = aioredis.from_url(redis_url, encoding="utf-8", decode_responses=True)
    FastAPICache.init(RedisBackend(redis), prefix="svc-demo-cache")


async def close_cache() -> None:
  # FastAPICache 无统一 close；可在 lifespan 里关闭 redis 连接池
    pass

svc_demo/main.py lifespan 片段：

from contextlib import asynccontextmanager
from svc_demo.core.cache import init_cache
from svc_demo.core.config import settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_cache(settings.REDIS_URL)
    yield
    # teardown

缓存商品列表

svc_demo/api/v1/products.py：

from fastapi import APIRouter, Depends, Query
from fastapi_cache.decorator import cache
from sqlalchemy.ext.asyncio import AsyncSession

from svc_demo.core.deps import get_db
from svc_demo.schemas.product import ProductListOut
from svc_demo.services.product import list_published_products

router = APIRouter(prefix="/products", tags=["products"])


@router.get("", response_model=ProductListOut)
@cache(expire=300)  # 5 分钟 TTL
async def get_products(
    page: int = Query(1, ge=1),
    per_page: int = Query(10, ge=1, le=50),
    q: str | None = None,
    db: AsyncSession = Depends(get_db),
):
    items, total = await list_published_products(db, page, per_page, q)
    return {
        "code": 0,
        "message": "ok",
        "data": items,
        "pagination": {
            "page": page,
            "per_page": per_page,
            "total": total,
            "pages": (total + per_page - 1) // per_page if total else 0,
        },
    }

注意：@cache 默认按完整 URL + 查询参数生成 key；page=1&per_page=10 与 page=2 互不冲突。

自定义 key_builder（选修）

当需要按业务维度缓存（忽略无关参数）：

from fastapi_cache import FastAPICache


def product_list_key_builder(
    func,
    namespace: str = "",
    request=None,
    response=None,
    *args,
    **kwargs,
):
    page = kwargs.get("page", 1)
    per_page = kwargs.get("per_page", 10)
    q = kwargs.get("q") or ""
    return f"{namespace}:products:{page}:{per_page}:{q}"


@router.get("/hot")
@cache(expire=60, key_builder=product_list_key_builder)
async def get_hot_products(db: AsyncSession = Depends(get_db)):
    ...

11.5 缓存失效：写后删除

创建/更新/下架商品后必须清理相关 key，否则用户看到过期数据。

svc_demo/services/product.py：

from redis import asyncio as aioredis
from svc_demo.core.config import settings


async def invalidate_product_cache() -> None:
    """删除 svc-demo-cache 前缀下所有商品列表缓存（教学简化版）。"""
    redis = aioredis.from_url(settings.REDIS_URL, decode_responses=True)
    cursor = 0
    prefix = "svc-demo-cache:*products*"
    while True:
        cursor, keys = await redis.scan(cursor, match=prefix, count=100)
        if keys:
            await redis.delete(*keys)
        if cursor == 0:
            break
    await redis.aclose()


async def create_product(db: AsyncSession, payload: ProductCreate) -> Product:
    product = Product(**payload.model_dump())
    db.add(product)
    await db.commit()
    await db.refresh(product)
    await invalidate_product_cache()
    return product

教学 MVP 推荐：写后删 + 短 TTL（60～300s）双保险。

11.6 aioredis 手写缓存（理解原理）

不依赖装饰器时，典型读路径：