Redis 实战教程

Redis 实战基础篇

1.1 缓存应用（Cache）

📌 缓存读写流程：Cache Aside 模式（旁路缓存）

流程简介：

• 读数据时，先查缓存，缓存未命中再查数据库，并写入缓存。
• 写数据时，先更新数据库，再删除缓存。

import redis
import time
from typing import Optional

r = redis.Redis(host='localhost', port=6379, db=0)

def get_from_db(user_id: str) -> dict:
    """模拟从数据库读取数据"""
    time.sleep(1)  # 模拟延迟
    return {"user_id": user_id, "username": f"user_{user_id}", "email": f"user{user_id}@test.com"}

def get_user(user_id: str) -> dict:
    key = f"user:{user_id}"
    user = r.get(key)
    if user:
        print("✅ 缓存命中")
        return eval(user)
    else:
        print("❌ 缓存未命中，从数据库查询")
        user = get_from_db(user_id)
        r.setex(key, 300, str(user))  # 缓存5分钟
        return user

def update_user(user_id: str, new_data: dict):
    # 先更新数据库（此处略）
    # 删除缓存
    r.delete(f"user:{user_id}")

---

⚠️ 缓存三大问题与应对策略

1️⃣ 缓存击穿（一个热点key突然失效）

• 方案：加互斥锁防止高并发同时打到DB，或设置 永不过期 + 后台定时刷新

def get_user_with_lock(user_id: str) -> dict:
    key = f"user:{user_id}"
    user = r.get(key)
    if user:
        return eval(user)
    
    lock_key = f"lock:{user_id}"
    if r.set(lock_key, "1", nx=True, ex=5):  # 分布式锁
        user = get_from_db(user_id)
        r.setex(key, 300, str(user))
        r.delete(lock_key)
        return user
    else:
        time.sleep(0.1)
        return get_user_with_lock(user_id)

2️⃣ 缓存雪崩（大量key同时过期）

• 方案：设置过期时间随机值，错峰缓存失效

import random

r.setex("key", 300 + random.randint(0, 60), "value")

3️⃣ 缓存穿透（请求不存在的数据）

• 方案：将空值也缓存起来（短时）

def get_user_safe(user_id: str) -> dict:
    key = f"user:{user_id}"
    user = r.get(key)
    if user:
        return eval(user)
    user = get_from_db(user_id)
    if not user:
        r.setex(key, 60, "None")
        return {}
    r.setex(key, 300, str(user))
    return user

---

1.2 Session 管理（用户状态）

🛠 Flask + Redis 实现 Session

from flask import Flask, request, jsonify
import uuid

app = Flask(__name__)
session_store = redis.Redis()

@app.route("/login", methods=["POST"])
def login():
    username = request.json.get("username")
    # 验证略...
    session_id = str(uuid.uuid4())
    session_store.setex(f"session:{session_id}", 3600, username)
    return jsonify({"session_id": session_id})

@app.route("/profile")
def profile():
    session_id = request.headers.get("Authorization")
    username = session_store.get(f"session:{session_id}")
    if not username:
        return jsonify({"error": "未登录"}), 401
    return jsonify({"username": username.decode()})

---

1.3 限流实战（防刷接口）

🚦 用 Redis 实现固定窗口限流（INCR）

def is_allowed(ip: str, limit=10) -> bool:
    key = f"rate:{ip}:{int(time.time() // 60)}"  # 每分钟一个key
    current = r.incr(key)
    if current == 1:
        r.expire(key, 60)
    return current <= limit

@app.route("/search")
def search():
    ip = request.remote_addr
    if not is_allowed(ip):
        return "限流中，请稍后再试", 429
    return "正常访问"

---

💡 进阶限流算法（补充理论）

策略	特点描述
固定窗口	简单易实现，但临界值易突增
滑动窗口	精度高但实现复杂（Lua脚本处理）
漏桶（Leaky）	匀速出水，限制处理速率
令牌桶（Token）	控制速率并支持突发

---

1.4 排行榜实战（ZSET）

🏆 构建一个游戏积分排行榜

def update_score(user_id: str, score: int):
    r.zadd("leaderboard", {user_id: score})

def get_top_n(n: int = 10):
    return r.zrevrange("leaderboard", 0, n - 1, withscores=True)

@app.route("/leaderboard")
def leaderboard():
    top = get_top_n()
    return jsonify(top)

---

🔍 多维排行榜（按时间、类别）

def update_score_by_day(user_id: str, score: int):
    day = time.strftime("%Y%m%d")
    r.zadd(f"leaderboard:{day}", {user_id: score})

def get_daily_top(day: str, n=10):
    return r.zrevrange(f"leaderboard:{day}", 0, n - 1, withscores=True)

---

✅ 总结

模块	技术点	场景
缓存	Cache Aside、雪崩击穿穿透处理	接口数据加速、热点数据缓存
Session 管理	Redis 存储登录态 + 过期自动控制	单点登录、微服务统一认证
限流	INCR、滑动窗口、漏桶令牌桶	防刷、风控、接口保护
排行榜	ZSET 构建有序集合、按时间分组	电商榜单、游戏排行、打榜投票类活动