Redis 缓存架构实战

为什么需要缓存？

数据库是瓶颈，缓存是解药。合理设计缓存能让系统承载 10 倍流量。

缓存更新策略

1. Cache Aside（旁路缓存）⭐推荐

def get_user(user_id):
    # 先读缓存
    user = redis.get(f"user:{user_id}")
    if user:
        return json.loads(user)
    
    # 缓存未命中，读数据库
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    
    # 写入缓存
    redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    
    return user

优点：简单、可靠 缺点：首次请求慢

2. Read/Write Through

应用只与缓存交互，缓存负责与数据库同步。

优点：代码简洁 缺点：需要缓存支持

3. Write Behind

先写缓存，异步写入数据库。

优点：写性能极高 缺点：可能丢失数据

缓存穿透解决方案

问题：查询不存在的数据，缓存不命中，压力直达数据库

方案一：布隆过滤器

from pybloom_live import BloomFilter

bloom = BloomFilter(capacity=1000000, error_rate=0.001)

# 添加存在的 ID
for user in all_users:
    bloom.add(user.id)

# 查询前检查
if user_id not in bloom:
    return None  # 肯定不存在

方案二：缓存空值

user = redis.get(f"user:{user_id}")
if user == "NULL":
    return None  # 缓存了空值

缓存击穿解决方案

问题：热点 key 过期，大量请求同时打到数据库

方案一：互斥锁

def get_user(user_id):
    user = redis.get(f"user:{user_id}")
    if user:
        return json.loads(user)
    
    # 获取互斥锁
    lock_key = f"lock:user:{user_id}"
    if redis.set(lock_key, "1", nx=True, ex=10):
        try:
            user = db.query(...)
            redis.setex(f"user:{user_id}", 3600, json.dumps(user))
            return user
        finally:
            redis.delete(lock_key)
    else:
        # 等待后重试
        time.sleep(0.1)
        return get_user(user_id)

方案二：逻辑过期

# 数据中包含过期时间
{
    "data": {...},
    "expire_at": 1711084200
}

# 发现快过期时异步更新
if data["expire_at"] - time.time() < 300:
    asyncio.create_task(refresh_cache(user_id))

缓存雪崩解决方案

问题：大量 key 同时过期

方案：随机过期时间

# 基础过期时间 + 随机值
base_ttl = 3600
random_ttl = random.randint(0, 300)
redis.setex(key, base_ttl + random_ttl, value)

缓存数据结构选择

场景	推荐结构
简单 KV	String
计数器	String (INCR)
排行榜	ZSet
消息队列	List
会话存储	Hash
位运算	Bitmap

监控与告警

# 查看内存使用
INFO memory

# 查看命中率
INFO stats

# 慢查询监控
SLOWLOG GET 10

结语

缓存是艺术，平衡一致性、性能、成本才是关键！