夜雨聆风你的 API 有多慢？
同一个查询请求，用户点一次要 2 秒，再点一次还要 2 秒...数据库每次都被折腾一遍，服务器累，用户也烦。
缓存就是解决这个问题的大招。把常用数据"暂存"一份在内存里，下次直接拿来用，省去重复计算的麻烦。
今天咱们聊聊 fastapi-cache —— 一个让你几行代码就给接口"装缓存"的插件。
读完本文，你将学会：
- • 缓存的基本原理和使用场景
- • 用 fastapi-cache 实现接口级缓存
- • 多种缓存后端：内存、Redis、Memcached
- • 缓存失效策略和最佳实践
一、缓存是干什么的？
一句话解释
缓存就是把昂贵的计算结果"存个副本"，下次直接拿来用。
什么时候用缓存？
| 场景 | 为什么需要缓存 |
|---|---|
| 热门文章查询 | 同一篇文章被几百万人看，每次都查数据库太浪费 |
| 配置信息读取 | 系统配置几乎不变，但每个请求都要读 |
| 排行榜计算 | 排行榜数据实时性要求不高，但计算很复杂 |
| 用户信息 | 用户登录后信息很少变，但每次接口都要取 |
| 第三方 API 调用 | 外部接口慢且有调用限制，缓存能减少请求 |
缓存的代价
缓存不是银弹，用不好会有问题：
- • 数据不一致：缓存和数据库数据对不上
- • 内存占用：缓存太多会撑爆内存
- • 缓存穿透：查询不存在的数据，每次都打到数据库
所以：该缓存的才缓存，不该缓存的别瞎存。
二、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() 在中间，函数定义在最下面！
本期分享就到这里啦，祝君在测开之路越走越远，越走越顺。
