Redis 缓存和限流

一句话：Redis 是放在内存里的高速小仓库，常用来做缓存、验证码、登录状态、限流和临时任务状态。

本篇学完你会什么：知道什么时候该用 Redis，能给接口加简单缓存和限流，不会只停留在 set/get。

1. Redis 解决什么问题

数据库适合长期保存数据，但有些数据访问太频繁，每次都查数据库会慢。

Redis 像服务台旁边的小抽屉：

常用资料放抽屉里
需要时马上拿
过一会儿自动清掉

Redis 缓存小抽屉示意图

2. 缓存是什么

缓存就是把常用结果先存起来。

没有缓存：

请求 -> 后端 -> 数据库 -> 后端 -> 前端

有缓存：

请求 -> 后端 -> Redis
              ↓ 如果没有
             数据库

第一次慢一点，后面相同请求就快。

3. 常见使用场景

场景	Redis 适合吗	例子
热门文章列表	适合	首页排行榜
验证码	适合	5 分钟过期
登录 token 黑名单	适合	退出登录
限流计数	适合	1 分钟最多 60 次
长期订单数据	不适合单独放	应该进数据库

4. 安装和连接

Docker 启动：

docker run --name demo-redis -p 6379:6379 -d redis:7

Python 依赖：

pip install redis

简单测试：

from redis.asyncio import Redis

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

await redis.set("hello", "world", ex=60)
value = await redis.get("hello")

ex=60 表示 60 秒后自动过期。

5. FastAPI 里怎么接 Redis

建议放到 core/redis.py：

from redis.asyncio import Redis

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

async def get_redis() -> Redis:
    return redis_client

接口里使用：

from fastapi import Depends
from redis.asyncio import Redis

@router.get("/ping-redis")
async def ping_redis(redis: Redis = Depends(get_redis)):
    await redis.set("ping", "pong", ex=10)
    return {"value": await redis.get("ping")}

6. 接口缓存

比如文章详情：

@router.get("/articles/{article_id}")
async def get_article(article_id: int, redis: Redis = Depends(get_redis)):
    cache_key = f"article:{article_id}"

    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)

    article = await query_article_from_db(article_id)
    data = {"id": article.id, "title": article.title}

    await redis.set(cache_key, json.dumps(data), ex=300)
    return data

大白话：

先查 Redis。
有就直接返回。
没有再查数据库。
查完放 Redis，过 5 分钟失效。

7. 缓存失效

缓存最大的难点是：数据库变了，缓存也要变。

常见策略：

策略	做法
TTL 自动过期	设置 `ex=300`
修改后删除缓存	更新文章后 `delete(article:1)`
手动刷新	管理员点刷新

更新文章时：

await update_article_in_db(article_id, payload)
await redis.delete(f"article:{article_id}")

先保证正确，再追求更快。

8. 接口限流

限流就是防止某个用户或 IP 短时间请求太多。

async def check_rate_limit(redis: Redis, key: str, limit: int, seconds: int):
    current = await redis.incr(key)
    if current == 1:
        await redis.expire(key, seconds)
    if current > limit:
        raise HTTPException(status_code=429, detail="请求太频繁")

这个版本适合教学和小项目理解思路。高并发生产环境里，更稳的做法是用 Redis Lua 脚本或封装好的限流组件，把 incr 和 expire 做成一次原子操作，避免中间步骤失败导致计数没有过期时间。

使用：

@router.post("/send-code")
async def send_code(request: Request, redis: Redis = Depends(get_redis)):
    ip = request.client.host
    await check_rate_limit(redis, f"send-code:{ip}", limit=5, seconds=60)
    return {"msg": "验证码已发送"}

9. 常见坑

问题	原因	解决
缓存一直不更新	数据库改了但没删缓存	更新后删除缓存
Redis 挂了服务也挂	没做降级	重要接口不能完全依赖缓存
key 命名混乱	没有统一规则	用 `业务:对象:id`
缓存太久	TTL 过长	根据业务设置过期
限流误伤用户	只按 IP 限制	登录后可以按用户 id 限制

10. 检查清单

[ ] 这个数据是否真的需要缓存
[ ] key 命名是否清楚
[ ] 是否设置 TTL
[ ] 数据修改后是否删除缓存
[ ] Redis 失败时服务是否能降级
[ ] 限流维度是 IP 还是用户
[ ] 429 错误是否提示清楚

总结表

名词	大白话
Redis	内存高速小仓库
缓存	常用结果先存起来
TTL	自动过期时间
key	缓存名字
限流	控制请求频率
429	请求太频繁
降级	Redis 出问题时别让主服务一起崩

下一篇建议：大白话讲解——后端系统分层架构。