4.2 常见应用场景

本章把前面学的知识组装成 6 个真实业务场景。每个场景给出：用什么类型、完整思路、关键代码（伪代码 + Python 示例）。 这也是面试"Redis 在项目里怎么用的"的标准素材库。

场景一：缓存数据库查询结果（Cache-Aside 模式）⭐⭐⭐

最重要的场景，必须掌握。 用 String 存 JSON。

读流程：

1. 先查 Redis
2. 命中 → 直接返回
3. 未命中 → 查数据库 → 写入 Redis（带过期时间）→ 返回

写流程（更新数据时）：

1. 先更新数据库
2. 再删除 Redis 里对应的缓存 key（下次读时自然会重建）

def get_product(pid):
    key = f"product:{pid}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    product = db.query("SELECT ... WHERE id=%s", pid)
    if product:
        r.set(key, json.dumps(product), ex=3600)
    return product

def update_product(pid, data):
    db.update(...)          # 1. 先更新数据库
    r.delete(f"product:{pid}")   # 2. 再删缓存

💡 为什么是"删缓存"而不是"改缓存"？因为改缓存在并发下容易把旧值写回去，删掉让下次读自动重建最简单可靠。为什么先更新库再删缓存？这是不一致概率最低的顺序（面试深挖点，新手记结论）。

场景二：接口限流 ⭐⭐

用 String + incr + expire。限制每个用户每 60 秒最多请求 100 次：

def is_allowed(user_id, limit=100, window=60):
    key = f"rate:{user_id}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)   # 第一次请求时开启窗口
    return count <= limit

原理：60 秒内的请求共用一个计数 key，到期自动清零。这是"固定窗口"算法，简单够用；更精确的"滑动窗口"用 ZSet 实现（进阶）。

场景三：排行榜 ⭐⭐

用 ZSet，前面章节已详细讲过，这里给出完整代码：

def add_score(player, score):
    r.zincrby("rank:game", score, player)

def top_n(n=10):
    # 返回 [(玩家, 分数), ...]，从高到低
    return r.zrevrange("rank:game", 0, n - 1, withscores=True)

def my_rank(player):
    rank = r.zrevrank("rank:game", player)
    return rank + 1 if rank is not None else None   # 转成从 1 开始

场景四：分布式锁 ⭐⭐⭐（高频面试题）

问题：定时任务部署在 3 台服务器上，如何保证同一时刻只有一台在执行？

方案：大家抢同一个 key，谁 SET NX 成功谁执行：

import uuid

def acquire_lock(lock_name, expire=30):
    token = str(uuid.uuid4())            # 唯一标识，标记锁是"我"加的
    ok = r.set(f"lock:{lock_name}", token, nx=True, ex=expire)
    return token if ok else None

def release_lock(lock_name, token):
    # 必须用 Lua：判断是自己的锁才能删，判断+删除要原子
    lua = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(lua, 1, f"lock:{lock_name}", token)

三个关键设计（面试就问这些）：

1. 为什么要设过期时间（EX）？ 持锁的服务器崩了，锁 30 秒后自动释放，不会死锁
2. 为什么 value 存随机 token？ 防止误删别人的锁：A 的锁过期后 B 拿到锁，A 恢复后如果直接 del 就把 B 的锁删了
3. 为什么释放锁要用 Lua？ "判断是我的锁"和"删除"必须原子完成，否则判断完的瞬间锁刚好过期易主，还是会误删

💡 生产环境建议直接用成熟库：Java 用 Redisson，Python 用 redis-py 自带的 Lock。原理就是上面这套 + 自动续期（看门狗）。

场景五：分布式 Session ⭐⭐

问题：用户在服务器 A 登录，下一个请求被负载均衡到服务器 B，B 不认识他。

方案：登录状态不存服务器本地，统一存 Redis：

def login(user):
    session_id = str(uuid.uuid4())
    r.set(f"session:{session_id}", json.dumps({"uid": user.id}), ex=7 * 86400)
    return session_id     # 通过 Cookie / Header 返回给浏览器

def get_current_user(session_id):
    data = r.get(f"session:{session_id}")
    return json.loads(data) if data else None   # None = 未登录或已过期

Token/JWT 方案里的"黑名单"、验证码、临时授权码，都是同样的套路。

场景六：秒杀库存预扣 ⭐⭐（进阶）

问题：1 万人抢 100 件商品，直接打到数据库必挂，还可能超卖。

方案：库存先加载进 Redis，用 Lua 原子扣减，抢到的人再走后续下单流程：

DEDUCT_LUA = """
local stock = tonumber(redis.call('get', KEYS[1]))
if stock and stock > 0 then
    return redis.call('decr', KEYS[1])
else
    return -1
end
"""

def seckill(product_id):
    result = r.eval(DEDUCT_LUA, 1, f"seckill:stock:{product_id}")
    if result >= 0:
        send_to_queue(product_id)   # 进消息队列异步创建订单
        return "抢购成功"
    return "已售罄"

Redis 单机能扛 10 万+ QPS，把 99% 的"已售罄"请求挡在数据库之外。

场景速查表

场景	数据类型	核心命令/技术
缓存	String	set ... EX + 先更新库再删缓存
限流	String	incr + expire
排行榜	ZSet	zincrby + zrevrange
分布式锁	String	set NX EX + Lua 释放
Session	String	set ... EX，sessionId 做 key
秒杀扣库存	String	Lua 原子扣减
购物车	Hash	hset / hincrby
点赞	Set	sadd / srem / sismember
签到	Bitmap	setbit / bitcount
UV 统计	HyperLogLog	pfadd / pfcount

---

下一章：缓存用多了必然遇到的三大问题 👉 4.3 缓存三大问题：穿透、击穿、雪崩