FastAPI缓存提速实战：手把手教你用Redis为接口注入“记忆”

你的FastAPI接口是不是在高并发下越来越慢，数据库频频告警？

一个案例，一个核心查询接口，在日活仅5万时，平均响应时间就飙升到了1.2秒。排查后发现，超过80%的请求都在重复查询数据库里那几条几乎不变的热点数据。在引入Redis缓存后，这个接口的平均响应时间直接降到了0.2秒以内，数据库负载下降了70%。这，就是缓存的魔力。

今天，我们就来聊聊如何为你FastAPI项目装上Redis这个“高速缓存”，让它拥有“记忆”，不再每次都傻傻地重复劳动。

📖 本文你将学到：

🎯 1. Redis是什么？为什么它是缓存的首选？

🎯 2. 如何快速安装、配置Redis。

🎯 3. 必须掌握的Redis核心命令。

🎯 4. 编写一个通用的FastAPI缓存装饰器，一劳永逸。

🔧 第一部分：问题与背景 - 为什么需要缓存？

想象一下，你是餐厅（你的Web服务）的服务员（API接口）。每次客人（客户端）点一份“今日特色菜”（热门数据），你都非得跑回后厨（数据库）问厨师一遍，尽管这道菜一天都不会变。结果就是，你累趴了，后厨也被你问烦了，客人还嫌上菜慢。

缓存，就像是你在前厅放了个小本子（Redis）。第一次有客人点“今日特色菜”，你去后厨问了，然后把菜名和价格记在本子上。接下来再有客人点，你直接看一眼本子就告诉他，速度快了十倍。只有当特色菜更换了（数据变更），你才需要去更新小本子。

在技术层面，缓存主要解决两个问题：1. 提升数据读取速度（内存远快于磁盘/网络）；2. 减轻后端数据库压力。

⚙️ 第二部分：Redis核心与安装配置

🎯 Redis是什么？

Redis是一个开源、基于内存、可持久化的键值对(Key-Value)存储系统。它支持多种数据结构（字符串、哈希、列表、集合等），性能极高，常被用作数据库、缓存和消息中间件。对于缓存场景，我们主要看中它：内存存储速度极快、数据结构丰富、支持设置过期时间。

🎯 安装Redis（各平台通用步骤）

1. macOS (使用Homebrew):

- 打开终端，执行：brew install redis

- 启动服务：brew services start redis

2. Linux (以Ubuntu为例):

- 更新包管理器：sudo apt update

- 安装Redis：sudo apt install redis-server

- 启动服务：sudo systemctl start redis

3. Windows:

官方不支持Windows原生安装，但可以通过：

- 使用WSL2（推荐，在WSL的Ubuntu中按Linux方法安装）。

- 或下载微软维护的旧版本Windows移植版（不推荐用于生产）。

安装完成后，在终端输入redis-cli ping，如果返回PONG，恭喜你，Redis服务已成功运行！

🎯 你必须掌握的5个Redis缓存核心命令

1. SET：设置键值对

/* by 01022.hk - online tools website : 01022.hk/zh/pinyin.html */ SET user:1001 '{"name": "Alice", "age": 30}' EX 60 # 键：user:1001， 值：JSON字符串， EX 60表示60秒后过期

2. GET：获取键对应的值

/* by 01022.hk - online tools website : 01022.hk/zh/pinyin.html */ GET user:1001

3. EXISTS：检查键是否存在

EXISTS user:1001 # 返回1(存在)或0(不存在)

4. DEL：删除一个或多个键

DEL user:1001 user:1002

5. TTL：查看键的剩余生存时间(秒)

TTL user:1001 # 返回剩余秒数，-1表示永不过期，-2表示键不存在

🚀 第三部分：FastAPI整合Redis实战演示

理论说再多，不如一行代码。让我们开始实战，构建一个带缓存的FastAPI应用。

🎯 第一步：安装依赖

pip install fastapi uvicorn redis python-dotenv

🎯 第二步：项目结构与配置

your_project/ ├── main.py ├── cache.py ├── .env └── requirements.txt

在.env文件中配置Redis连接：

REDIS_HOST=localhost REDIS_PORT=6379 REDIS_DB=0 REDIS_PASSWORD= # 默认无密码，生产环境一定要设！ CACHE_DEFAULT_TTL=3600 # 默认缓存过期时间1小时

🎯 第三步：创建Redis连接与缓存工具类 (cache.py)

import redis.asyncio as redis # 使用异步客户端 import json from functools import wraps from typing import Any, Optional import os from dotenv import load_dotenv load_dotenv() class RedisCache: def __init__(self): self.redis_client: Optional[redis.Redis] = None self.default_ttl = int(os.getenv(“CACHE_DEFAULT_TTL”, 3600)) async def connect(self): """连接Redis""" if not self.redis_client: self.redis_client = redis.Redis( host=os.getenv(“REDIS_HOST”, “localhost”), port=int(os.getenv(“REDIS_PORT”, 6379)), db=int(os.getenv(“REDIS_DB”, 0)), password=os.getenv(“REDIS_PASSWORD”), decode_responses=True # 自动解码返回字符串 ) return self.redis_client async def disconnect(self): """关闭连接""" if self.redis_client: await self.redis_client.close() def cache_key(self, func_name: str, *args, **kwargs) -> str: """生成唯一的缓存键""" # 简单示例：将函数名和参数序列化后拼接 arg_str = “_”.join([str(arg) for arg in args]) kwarg_str = “_”.join([f“{k}_{v}” for k, v in sorted(kwargs.items())]) return f“fastapi_cache:{func_name}:{arg_str}:{kwarg_str}”.strip(“:”) async def get(self, key: str) -> Any: """从缓存获取数据""" if not self.redis_client: await self.connect() data = await self.redis_client.get(key) if data: try: return json.loads(data) # 反序列化JSON except json.JSONDecodeError: return data # 如果不是JSON，返回原始字符串 return None async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool: """设置缓存""" if not self.redis_client: await self.connect() if isinstance(value, (dict, list)): value = json.dumps(value) # 序列化复杂对象 expire_time = ttl if ttl is not None else self.default_ttl return await self.redis_client.setex(key, expire_time, value) async def delete(self, key: str) -> int: """删除缓存""" if not self.redis_client: await self.connect() return await self.redis_client.delete(key) # 全局缓存实例 cache = RedisCache() def cached(ttl: Optional[int] = None): """缓存装饰器：可复用于任何异步函数""" def decorator(func): @wraps(func) async def wrapper(*args, **kwargs): # 生成缓存键 key = cache.cache_key(func.__name__, *args, **kwargs) # 尝试从缓存获取 cached_result = await cache.get(key) if cached_result is not None: print(f“Cache HIT for key: {key}”) return cached_result # 缓存未命中，执行原函数 print(f“Cache MISS for key: {key}”) result = await func(*args, **kwargs) # 将结果存入缓存 await cache.set(key, result, ttl) return result return wrapper return decorator

🎯 第四步：在FastAPI应用中使用 (main.py)

from fastapi import FastAPI, Depends, HTTPException from cache import cache, cached import asyncio app = FastAPI(title=“FastAPI Redis缓存演示”) # 应用启动和关闭事件 @app.on_event(“startup”) async def startup_event(): await cache.connect() print(“✅ Redis connected”) @app.on_event(“shutdown”) async def shutdown_event(): await cache.disconnect() print(“👋 Redis disconnected”) # --- 模拟一个耗时的数据查询函数 --- async def fetch_user_data_from_db(user_id: int): """模拟从数据库查询用户数据（耗时操作）""" await asyncio.sleep(2) # 模拟2秒的IO延迟 return {“id”: user_id, “name”: f“用户_{user_id}”, “score”: user_id * 10} # --- 应用缓存的接口 --- @app.get(“/user/{user_id}”) @cached(ttl=30) # 为此接口单独设置30秒缓存 async def get_user(user_id: int): """获取用户信息（带缓存）""" data = await fetch_user_data_from_db(user_id) return {“source”: “database (cached later)”, “data”: data} @app.get(“/user/{user_id}/fresh”) async def get_user_fresh(user_id: int): """获取用户信息（强制查数据库，不缓存）""" data = await fetch_user_data_from_db(user_id) return {“source”: “database (fresh)”, “data”: data} @app.delete(“/cache/user/{user_id}”) async def delete_user_cache(user_id: int): """手动删除某个用户的缓存""" # 注意：这里需要模拟生成和接口一致的缓存键，实战中可能需要更复杂的键管理 key_pattern = f“fastapi_cache:get_user:{user_id}” deleted = await cache.delete(key_pattern) if deleted: return {“message”: f”Cache for user {user_id} deleted.”} raise HTTPException(status_code=404, detail=“Cache key not found”) if __name__ == “__main__”: import uvicorn uvicorn.run(“main:app”, host=“0.0.0.0”, port=8000, reload=True)

现在，运行python main.py并访问http://localhost:8000/docs查看自动生成的API文档。

测试效果：

- 首次访问/user/1，会等待约2秒，返回来源为database。

-30秒内再次访问/user/1，瞬间返回，来源数据来自缓存，控制台会打印Cache HIT。

- 访问/user/1/fresh则总是访问“数据库”。

- 调用DELETE /cache/user/1可以手动清除缓存。