在Web开发中，我们经常面临两个核心问题：如何快速响应请求，以及如何在多个请求间共享临时状态。FastAPI作为高性能Python Web框架，与Redis（一个内存数据存储系统）的组合能有效解决这两个问题。本文将通过简单示例，带你理解它们如何协同工作。
一、为什么需要FastAPI+Redis？¶
- FastAPI：提供了快速构建API的能力，支持异步请求处理，适合高并发场景。但即使FastAPI再快，对高频、重复的计算或数据查询（如用户信息、统计数据）仍需额外优化。
- Redis：高性能的内存数据库，支持键值对存储，能快速读写数据，且支持过期时间、分布式锁等特性。常用于缓存（减少重复计算）和临时状态管理（如会话、计数器）。
二、环境准备¶
首先安装必要的工具：
# 1. 安装FastAPI和Uvicorn（Web服务器）
pip install fastapi uvicorn
# 2. 安装Redis Python客户端
pip install redis
同时确保本地已安装并启动Redis服务（默认端口6379，无密码可直接连接）。验证连接：
redis-cli # 进入Redis命令行，执行ping，返回PONG则连接成功
三、Redis基础连接¶
在FastAPI中，我们可以通过工具函数统一管理Redis连接，避免重复创建连接：
from fastapi import FastAPI
import redis
import json
# 初始化Redis连接（生产环境建议用连接池，此处简化）
redis_client = redis.StrictRedis(
host="localhost", # Redis服务地址
port=6379, # 默认端口
db=0, # 数据库编号
decode_responses=True # 自动将结果转为字符串
)
app = FastAPI()
# 测试连接是否正常
@app.get("/redis-test")
async def redis_test():
redis_client.set("test_key", "Hello Redis!") # 存入键值对
return {"message": redis_client.get("test_key")} # 读取键值对
说明：
decode_responses=True确保Redis返回字符串而非字节，方便直接处理。
四、缓存：减少重复计算的利器¶
缓存的核心是将高频访问、低更新的数据（如固定配置、用户基本信息）存入Redis，下次请求直接从缓存读取，避免重复计算或数据库查询。
示例：假设我们有一个API需要“耗时计算”一个结果（如斐波那契数列，模拟复杂逻辑），每次请求都需重新计算，非常低效。用Redis缓存结果：
# 模拟“耗时计算”（实际可能是数据库查询或复杂算法）
def slow_calculate(n: int) -> int:
if n <= 1:
return n
return slow_calculate(n-1) + slow_calculate(n-2)
# FastAPI视图函数：带缓存逻辑
from fastapi import HTTPException
@app.get("/cache/calculate/{n}")
async def cached_calculate(n: int):
cache_key = f"fib:{n}" # 缓存键，用n区分不同计算结果
# 1. 先查缓存
cached_result = redis_client.get(cache_key)
if cached_result:
return {"n": n, "result": int(cached_result), "source": "cache"}
# 2. 缓存未命中，执行耗时计算
try:
result = slow_calculate(n)
except Exception as e:
raise HTTPException(status_code=500, detail="计算失败")
# 3. 存入缓存，设置过期时间（10分钟，避免数据永久占用）
redis_client.setex(cache_key, 600, str(result)) # setex：设置键值+过期时间
return {"n": n, "result": result, "source": "database"}
效果：首次请求
/cache/calculate/10会执行计算并缓存，后续请求直接从Redis读取，无需重复计算。
五、状态管理：跨请求共享临时数据¶
Redis也可用于存储临时状态（如用户会话、操作计数器），解决FastAPI多请求间的状态共享问题（内存变量仅在单进程内有效，Redis可跨进程/服务器共享）。
示例：统计用户访问次数（临时状态）：
@app.get("/counter/{user_id}")
async def user_counter(user_id: str):
cache_key = f"counter:{user_id}" # 状态键，关联用户ID
# 1. 从Redis读取当前访问次数
current_count = redis_client.get(cache_key)
if current_count:
current_count = int(current_count) + 1 # 次数+1
else:
current_count = 1 # 新用户，初始值为1
# 2. 存入Redis，设置过期时间（24小时，用户长时间未访问则自动失效）
redis_client.setex(cache_key, 86400, str(current_count))
return {"user_id": user_id, "访问次数": current_count}
效果：用户
user_id=123第一次访问返回访问次数=1，第二次访问返回访问次数=2，数据在Redis中持久化，即使服务重启也不会丢失。
六、进阶技巧：依赖注入优化代码¶
为避免重复编写Redis连接逻辑，可通过FastAPI的依赖注入统一管理Redis客户端：
from fastapi import Depends
# 定义Redis连接依赖
async def get_redis_client():
return redis.StrictRedis(host="localhost", port=6379, db=0, decode_responses=True)
# 视图函数中使用依赖
@app.get("/example")
async def example(redis_client=Depends(get_redis_client)):
redis_client.set("key", "value")
return {"data": redis_client.get("key")}
七、总结¶
- 缓存：核心是“高频数据→Redis存储→重复请求直接读取”，适合减少数据库/计算压力（如用户信息、统计数据）。
- 状态管理：核心是“跨请求共享临时数据→Redis键值对+过期时间”，适合会话、计数器、权限验证等场景。
通过FastAPI+Redis的组合，既能发挥FastAPI的异步高性能，又能利用Redis的内存速度优势，轻松解决Web开发中的“响应慢”和“状态共享”问题。
注意：生产环境需注意Redis的持久化配置（防止数据丢失）、连接池优化（避免频繁创建连接），以及键命名规范（如fib:{n}、counter:{user_id}）防止冲突。