南平市网站建设_网站建设公司_HTTPS_seo优化

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

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

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

📖 本文你将学到：

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

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

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

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

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

🎯 Redis是什么？

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

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

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

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

/* 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秒后过期

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

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

DEL user:1001 user:1002

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文档。

测试效果：

💡 第四部分：注意事项与进阶思考

---写在最后---
希望这份总结能帮你避开一些坑。如果觉得有用，不妨点个 赞👍 或 收藏⭐ 标记一下，方便随时回顾。也欢迎关注我，后续为你带来更多类似的实战解析。有任何疑问或想法，我们评论区见，一起交流开发中的各种心得与问题。