FastAPI 是一个现代、高性能的 Python Web 框架，广泛用于构建 RESTful API。而 Redis 作为一个高性能的内存数据库，常被用作缓存系统来提升应用程序的响应速度。

1. 什么是 FastAPI 和 Redis？

1.1 FastAPI 简介

FastAPI 是一个基于 Python 3.6+ 的 Web 框架，充分利用了类型注解和异步编程特性。它以高性能和易用性著称，特别适合构建需要快速响应的 API 服务。

1.2 Redis 简介

Redis（Remote Dictionary Server）是一个开源的键值存储数据库，支持多种数据结构（如字符串、哈希、列表等）。它主要运行在内存中，因此速度极快，常用于缓存、会话管理等场景。

2. 为什么在 FastAPI 中使用 Redis 缓存？

在实际应用中，某些数据（如数据库查询结果、API 响应等）可能需要重复计算或访问。通过将这些数据存储在 Redis 中，可以显著减少计算开销和响应时间。FastAPI 的异步特性与 Redis 的高性能天然契合，使其成为理想的组合。

3. 环境准备

在开始之前，需要安装必要的依赖：

3.1 安装 FastAPI 和相关库

pip install fastapi uvicorn

3.2 安装 Redis Python 客户端

pip install redis

3.3 安装并运行 Redis

• 在本地安装 Redis（Windows、Linux 或 macOS 均可），具体安装方式请参考 Redis 官网。
• 启动 Redis 服务：

redis-server

• 确保 Redis 默认运行在 localhost:6379。

4. FastAPI 中集成 Redis 的基本实现

以下是一个简单的 FastAPI 应用，展示如何使用 Redis 缓存数据。

4.1 基本代码结构

from fastapi import FastAPI
import redis
import json
import time

# 初始化 FastAPI 应用
app = FastAPI()

# 初始化 Redis 连接
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 模拟一个耗时操作
def expensive_computation():
    time.sleep(2)  # 模拟 2 秒的计算时间
    return {"message": "This is an expensive computation result"}

# 获取数据的端点
@app.get("/data")
async def get_data():
    # 检查缓存是否存在
    cache_key = "expensive_data"
    cached_data = redis_client.get(cache_key)
    
    if cached_data:
        # 如果缓存命中，直接返回缓存数据
        return json.loads(cached_data)
    
    # 如果缓存未命中，执行计算并存储到 Redis
    result = expensive_computation()
    redis_client.setex(cache_key, 60, json.dumps(result))  # 设置 60 秒过期时间
    return result

# 运行应用
# 使用命令：uvicorn main:app --reload

4.2 代码说明

1. Redis 连接：redis.Redis 创建了一个与本地 Redis 实例的连接，decode_responses=True 确保返回字符串而不是字节。
2. 缓存逻辑：
3. 使用 redis_client.get 检查缓存。
4. 如果缓存存在，直接返回缓存数据（需用 json.loads 反序列化）。
5. 如果缓存不存在，执行 expensive_computation，并用 redis_client.setex 将结果存入 Redis，设置 60 秒过期时间。
6. 异步支持：FastAPI 的端点使用 async def，与 Redis 的异步操作兼容。

4.3 测试

运行应用后，访问
http://127.0.0.1:8000/data：

• 第一次请求会延迟 2 秒（模拟计算时间）。
• 在接下来的 60 秒内，重复请求将立即返回缓存结果。

5. 高级示例：带参数的缓存

在实际场景中，API 可能需要根据参数返回不同数据。我们可以通过动态生成缓存键来实现。

5.1 代码实现

from fastapi import FastAPI, HTTPException
import redis
import json
import time

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 模拟数据库查询
def fetch_data_from_db(user_id: int):
    time.sleep(1)  # 模拟 1 秒的数据库查询
    if user_id == 1:
        return {"id": 1, "name": "Alice"}
    elif user_id == 2:
        return {"id": 2, "name": "Bob"}
    else:
        raise HTTPException(status_code=404, detail="User not found")

# 根据用户 ID 获取数据的端点
@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # 动态生成缓存键
    cache_key = f"user:{user_id}"
    cached_data = redis_client.get(cache_key)
    
    if cached_data:
        return json.loads(cached_data)
    
    # 未命中缓存，从“数据库”获取数据
    result = fetch_data_from_db(user_id)
    redis_client.setex(cache_key, 30, json.dumps(result))  # 缓存 30 秒
    return result

5.2 代码说明

• 动态缓存键：cache_key = f"user:{user_id}" 根据 user_id 生成唯一的缓存键。
• 错误处理：如果 fetch_data_from_db 抛出异常，FastAPI 会自动返回 HTTP 404 响应。
• 过期时间：缓存设置为 30 秒，适合动态数据场景。

5.3 测试

• 访问 http://127.0.0.1:8000/user/1：第一次请求延迟 1 秒，后续 30 秒内快速返回。
• 访问 http://127.0.0.1:8000/user/3：返回 404 错误。

6. 注意事项与优化

6.1 Redis 连接管理

在生产环境中，建议使用连接池管理 Redis 连接：

redis_client = redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)

6.2 缓存失效策略

• 使用 setex 设置过期时间，避免缓存占用过多内存。
• 根据业务需求调整过期时间，或实现手动失效（如 redis_client.delete(cache_key)）。

6.3 序列化与反序列化

复杂对象需要序列化为 JSON 存储，使用 json.dumps 和 json.loads 是常见做法。

6.4 异步 Redis 客户端

如果需要完全异步操作，可以使用 aioredis：

pip install aioredis

然后修改代码为：

import aioredis

redis_client = await aioredis.create_redis_pool("redis://localhost:6379/0")
await redis_client.setex(cache_key, 60, json.dumps(result))

7. 总结

通过将 Redis 集成到 FastAPI 中，我们可以显著提升 API 的性能，尤其是在处理耗时操作或频繁请求时。基本实现适用于简单场景，而带参数的缓存展示了更灵活的应用方式。在生产环境中，还需考虑连接管理、缓存策略和异步优化等因素。

感谢你看到这里，如果觉得文章对你有所收获，请在文末为我点个【赞】+【推荐】，或者【转发】给身边更多有需要的人看，你的点赞就是对我莫大的支持与动力!