使用 fastapi-cache 插件实现缓存机制提高性能

你的 API 有多慢？

同一个查询请求，用户点一次要 2 秒，再点一次还要 2 秒...数据库每次都被折腾一遍，服务器累，用户也烦。

缓存就是解决这个问题的大招。把常用数据"暂存"一份在内存里，下次直接拿来用，省去重复计算的麻烦。

今天咱们聊聊 fastapi-cache —— 一个让你几行代码就给接口"装缓存"的插件。

读完本文，你将学会：

• • 缓存的基本原理和使用场景
• • 用 fastapi-cache 实现接口级缓存
• • 多种缓存后端：内存、Redis、Memcached
• • 缓存失效策略和最佳实践

一、缓存是干什么的？

一句话解释

缓存就是把昂贵的计算结果"存个副本"，下次直接拿来用。

什么时候用缓存？

缓存的代价

缓存不是银弹，用不好会有问题：

• • 数据不一致：缓存和数据库数据对不上
• • 内存占用：缓存太多会撑爆内存
• • 缓存穿透：查询不存在的数据，每次都打到数据库

所以：该缓存的才缓存，不该缓存的别瞎存。

二、fastapi-cache 基础用法

项目结构

cache-demo/
├── main.py          # 入口文件
├── redis_client.py  # Redis 连接
└── requirements.txt # 依赖

依赖安装

fastapi==0.104.1
uvicorn[standard]==0.24.0
fastapi-cache2==0.2.1
redis==5.0.1

错误写法（新手常见）

from fastapi import FastAPI

app = FastAPI()

# ❌ 错误：自己写缓存逻辑
_cache = {}

@app.get("/items")
async def get_items():
    if "items" in _cache:
        return _cache["items"]  # 直接返回缓存
    
    # 模拟查询数据库
    result = await fetch_from_db()
    _cache["items"] = result  # 存入缓存
    return result

# ❌ 错误：没有缓存过期时间
# 数据永远不过期，数据库更新了用户还看到旧数据

问题在哪？

• • 自己管理缓存逻辑，代码杂乱
• • 没有过期策略，数据一致性难保证
• • 单机内存缓存，无法共享

正确写法（基础版本）

import redis.asyncio as redis
from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

app = FastAPI(title="FastAPI 缓存示例")

@app.on_event("startup")
async def startup():
    # ✅ 初始化 Redis 缓存后端
    redis_instance = redis.from_url(
        "redis://localhost:6379/0",
        encoding="utf-8",
        decode_responses=True
    )
    FastAPICache.init(RedisBackend(redis_instance), prefix="fastapi-cache")

# ✅ 使用装饰器缓存，过期时间 60 秒
@app.get("/items")
@cache(expire=60)
async def get_items():
    """获取商品列表 - 缓存 60 秒"""
    # 模拟耗时查询
    import asyncio
    await asyncio.sleep(2)  # 模拟数据库查询 2 秒
    
    return {"items": ["苹果", "香蕉", "橙子"], "source": "database"}

@app.get("/item/{item_id}")
@cache(expire=30)
async def get_item(item_id: int):
    """获取单个商品 - 缓存 30 秒"""
    await asyncio.sleep(1)
    return {"id": item_id, "name": f"商品-{item_id}", "price": 99.9}

优化写法（生产版本）

import logging
from typing import Optional
import redis.asyncio as redis
from fastapi import FastAPI, HTTPException, Request
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache
from fastapi_cache.coder import JsonCoder

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 配置参数
class Config:
    """应用配置"""
    REDIS_URL = "redis://localhost:6379/0"
    HOST = "0.0.0.0"
    PORT = 8000
    DEFAULT_CACHE_EXPIRE = 300  # 默认缓存 5 分钟

app = FastAPI(title="FastAPI 缓存实战", version="1.0.0")

@app.on_event("startup")
async def startup():
    """启动时初始化 Redis 缓存"""
    try:
        redis_instance = redis.from_url(
            Config.REDIS_URL,
            encoding="utf-8",
            decode_responses=True
        )
        FastAPICache.init(
            RedisBackend(redis_instance),
            prefix="fastapi-cache",
            coder=JsonCoder  # 使用 JSON 序列化
        )
        logger.info("✅ 缓存初始化成功")
    except Exception as e:
        logger.error(f"❌ Redis 缓存初始化失败: {e}")
        raise

@app.on_event("shutdown")
async def shutdown():
    """关闭时清理资源"""
    logger.info("✅ 缓存已关闭")

# ==================== 缓存装饰器工厂 ====================

def custom_cache(expire: int = Config.DEFAULT_CACHE_EXPIRE):
    """
    自定义缓存装饰器工厂
    
    统一配置缓存参数，避免到处写重复代码
    """
    return cache(
        expire=expire,
        coder=JsonCoder,
        key_builder=None  # 使用默认键生成器
    )

# ==================== 路由定义 ====================

@app.get("/")
@custom_cache(expire=60)
async def root():
    """首页 - 缓存 1 分钟"""
    return {"message": "FastAPI 缓存演示", "timestamp": "2024-01-01T00:00:00"}

@app.get("/products")
@custom_cache(expire=300)  # 缓存 5 分钟
async def get_products():
    """
    获取商品列表 - 缓存 5 分钟
    
    商品列表变化不频繁，适合长缓存
    """
    try:
        logger.info("📊 从数据库查询商品列表...")
        import asyncio
        await asyncio.sleep(2)  # 模拟数据库查询
        
        products = [
            {"id": 1, "name": "iPhone 15", "price": 5999},
            {"id": 2, "name": "MacBook Pro", "price": 14999},
            {"id": 3, "name": "AirPods Pro", "price": 1999},
        ]
        
        return {"code": 0, "data": products, "msg": "success"}
    except Exception as e:
        logger.error(f"查询商品列表失败: {e}")
        raise HTTPException(500, "服务器内部错误")

@app.get("/product/{product_id}")
@custom_cache(expire=60)  # 缓存 1 分钟
async def get_product(product_id: int):
    """
    获取单个商品 - 缓存 1 分钟
    
    单个商品可能被频繁查看
    """
    try:
        logger.info(f"📊 从数据库查询商品 {product_id}...")
        await asyncio.sleep(1)
        
        return {
            "code": 0,
            "data": {
                "id": product_id,
                "name": f"商品-{product_id}",
                "price": 99.9 * product_id,
                "stock": 100
            }
        }
    except Exception as e:
        logger.error(f"查询商品失败: {e}")
        raise HTTPException(500, "查询失败")

@app.get("/hot-articles")
@custom_cache(expire=600)  # 缓存 10 分钟
async def get_hot_articles():
    """
    获取热门文章 - 缓存 10 分钟
    
    排行榜计算复杂，数据实时性要求不高
    """
    try:
        logger.info("🔥 计算热门文章排行榜...")
        await asyncio.sleep(3)  # 模拟复杂计算
        
        articles = [
            {"id": 1, "title": "FastAPI 入门指南", "views": 10000},
            {"id": 2, "title": "Python 异步编程", "views": 8500},
            {"id": 3, "title": "Redis 缓存实战", "views": 7200},
        ]
        
        return {"code": 0, "data": articles, "msg": "success"}
    except Exception as e:
        logger.error(f"计算排行榜失败: {e}")
        raise HTTPException(500, "排行榜计算失败")

# ==================== 手动缓存控制 ====================

@app.post("/product/{product_id}/update")
async def update_product(product_id: int):
    """
    更新商品 - 清除相关缓存
    
    数据更新后，缓存要失效，否则用户看到旧数据
    """
    try:
        # 1. 更新数据库
        logger.info(f"📝 更新商品 {product_id}...")
        await asyncio.sleep(1)
        
        # 2. 清除该商品的缓存
        from fastapi_cache import FastAPICache
        backend = FastAPICache.get_backend()
        
        # 清除特定商品缓存
        await backend.clear(f"fastapi-cache:get_product:{product_id}")
        
        # 3. 清除商品列表缓存（列表也变了）
        await backend.clear("fastapi-cache:get_products")
        
        logger.info("✅ 缓存已清除")
        
        return {"code": 0, "msg": "更新成功，缓存已刷新"}
    except Exception as e:
        logger.error(f"更新失败: {e}")
        raise HTTPException(500, "更新失败")

@app.get("/health")
async def health_check():
    """健康检查 - 不缓存"""
    return {"status": "healthy", "cache": "enabled"}

if __name__ == "__main__":
    import uvicorn
    import asyncio
    uvicorn.run(app, host=Config.HOST, port=Config.PORT)

三、缓存键自定义

默认缓存键的问题

默认的缓存键是 prefix:函数名:参数，但如果参数是对象或包含敏感信息，就不合适了。

自定义缓存键

from fastapi import Request
from fastapi_cache.decorator import cache

async def custom_key_builder(
    func,
    namespace: Optional[str] = "",
    request: Optional[Request] = None,
    response=None,
    *args,
    **kwargs
):
    """
    自定义缓存键生成器
    
    规则：
    1. 使用用户 ID 作为前缀（隔离不同用户的数据）
    2. 使用接口路径作为键
    3. 忽略敏感参数（如 token）
    """
    user_id = request.headers.get("X-User-ID", "anonymous")
    path = request.url.path
    query = str(sorted(request.query_params.items()))
    
    return f"{namespace}:{user_id}:{path}:{query}"

@app.get("/user/profile")
@cache(expire=300, key_builder=custom_key_builder)
async def get_user_profile(request: Request):
    """获取用户资料 - 按用户隔离缓存"""
    user_id = request.headers.get("X-User-ID")
    return {"user_id": user_id, "name": "张三", "level": "VIP"}

四、多种缓存后端

1. 内存缓存（开发测试用）

from fastapi_cache.backends.inmemory import InMemoryBackend

@app.on_event("startup")
async def startup():
    # ✅ 内存缓存，重启即清空，适合开发测试
    FastAPICache.init(InMemoryBackend(), prefix="fastapi-cache")

2. Redis 缓存（生产推荐）

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

@app.on_event("startup")
async def startup():
    redis_instance = redis.from_url("redis://localhost:6379/0")
    FastAPICache.init(RedisBackend(redis_instance), prefix="fastapi-cache")

3. Memcached 缓存

import memcache
from fastapi_cache.backends.memcached import MemcachedBackend

@app.on_event("startup")
async def startup():
    client = memcache.Client(["127.0.0.1:11211"])
    FastAPICache.init(MemcachedBackend(client), prefix="fastapi-cache")

五、高级缓存策略

策略 1：条件缓存

不是所有请求都值得缓存：

from fastapi import Request
from fastapi_cache.decorator import cache

@app.get("/search")
@cache(expire=60)
async def search(request: Request, keyword: str):
    """
    搜索接口 - 智能缓存
    
    热门关键词缓存，冷门关键词不缓存
    """
    # 检查是否是热门关键词（可以用布隆过滤器或 Set）
    hot_keywords = {"FastAPI", "Python", "Redis", "Docker"}
    
    if keyword in hot_keywords:
        # 热门词，缓存结果
        return await search_database(keyword)
    else:
        # 冷门词，直接查库，不缓存
        return await search_database(keyword)

策略 2：缓存预热

系统启动时提前把热点数据放入缓存：

@app.on_event("startup")
async def startup():
    # 初始化 Redis
    redis_instance = redis.from_url("redis://localhost:6379/0")
    FastAPICache.init(RedisBackend(redis_instance), prefix="fastapi-cache")
    
    # ✅ 预热缓存：把热门数据提前加载
    await warm_up_cache()

async def warm_up_cache():
    """缓存预热"""
    logger.info("🔥 开始缓存预热...")
    
    # 预热商品列表
    products = await fetch_products_from_db()
    await cache_products(products)
    
    # 预热配置信息
    config = await fetch_config_from_db()
    await cache_config(config)
    
    logger.info("✅ 缓存预热完成")

策略 3：多级缓存

内存 + Redis 组合，兼顾速度和容量：

import asyncio
from typing import Any, Optional

class MultiLevelCache:
    """
    多级缓存：L1 内存 + L2 Redis
    
    先查内存，没有再查 Redis，都没有才查数据库
    """
    def __init__(self):
        self._memory = {}  # L1 内存缓存
        self._redis = None  # L2 Redis 缓存
    
    async def get(self, key: str) -> Optional[Any]:
        # 1. 查内存
        if key in self._memory:
            return self._memory[key]
        
        # 2. 查 Redis
        if self._redis:
            value = await self._redis.get(key)
            if value:
                # 回填内存
                self._memory[key] = value
                return value
        
        return None
    
    async def set(self, key: str, value: Any, expire: int = 300):
        # 同时写入内存和 Redis
        self._memory[key] = value
        if self._redis:
            await self._redis.setex(key, expire, value)

六、缓存失效和一致性

1. 主动失效

from fastapi_cache import FastAPICache

@app.delete("/product/{product_id}")
async def delete_product(product_id: int):
    """删除商品后清除缓存"""
    # 删除数据库记录
    await db.delete(product_id)
    
    # 清除缓存
    backend = FastAPICache.get_backend()
    await backend.clear(f"fastapi-cache:get_product:{product_id}")
    await backend.clear("fastapi-cache:get_products")
    
    return {"code": 0, "msg": "删除成功"}

2. 定时失效

from fastapi_cache.decorator import cache

@app.get("/daily-report")
@cache(expire=3600)  # 1 小时后自动失效
async def get_daily_report():
    """日报数据 - 每小时更新一次"""
    return await generate_report()

3. 监听数据库变更（高级）

# 使用数据库触发器或消息队列
# 当数据变更时，发送消息清除缓存

async def on_database_change(table: str, record_id: int):
    """数据库变更监听"""
    if table == "products":
        backend = FastAPICache.get_backend()
        await backend.clear(f"fastapi-cache:get_product:{record_id}")
        await backend.clear("fastapi-cache:get_products")

七、测试缓存效果

对比测试

import time
import requests

def test_cache_performance():
    """测试缓存前后性能对比"""
    url = "http://localhost:8000/products"
    
    # 第一次请求（无缓存，查数据库）
    start = time.time()
    r1 = requests.get(url)
    time_no_cache = time.time() - start
    
    # 第二次请求（有缓存，直接返回）
    start = time.time()
    r2 = requests.get(url)
    time_with_cache = time.time() - start
    
    print(f"无缓存: {time_no_cache:.2f} 秒")
    print(f"有缓存: {time_with_cache:.2f} 秒")
    print(f"提速: {time_no_cache / time_with_cache:.1f}x")

# 输出示例：
# 无缓存: 2.05 秒
# 有缓存: 0.01 秒
# 提速: 205.0x

总结

今天我们一起学习了 FastAPI 缓存实战：

• • 缓存的本质：把昂贵的计算结果存副本，下次直接用
• • fastapi-cache 核心：@cache 装饰器 + Redis 后端，几行代码搞定
• • 缓存粒度：接口级缓存，支持自定义过期时间和缓存键
• • 进阶策略：条件缓存、缓存预热、多级缓存
• • 数据一致性：更新数据时主动清除缓存，保证数据新鲜

缓存用得好，接口响应从秒级降到毫秒级，用户体验直线上升。

每日踩一坑：

@cache 装饰器必须放在 @app.get() 下面！

正确的写法是 @app.get() 在最上面，@cache() 在中间，函数定义在最下面！

本期分享就到这里啦，祝君在测开之路越走越远，越走越顺。