Redis 缓存应用

Redis是高性能缓存系统的首选，合理使用缓存可大幅降低数据库压力，提升系统响应速度。

缓存架构

缓存作用

- 减少数据库访问
- 降低响应延迟
- 提高系统吞吐
- 节省数据库资源

缓存位置

应用架构：
客户端 → 应用服务 → Redis缓存 → 数据库

读流程：
1. 查缓存
2. 缓存存在 → 直接返回
3. 缓存不存在 → 查数据库 → 写缓存 → 返回

缓存策略

Cache-Aside（旁路缓存）

# 读流程
def get_data(key):
    # 1. 先查缓存
    data = redis.get(key)
    if data:
        return data

    # 2. 缓存不存在，查数据库
    data = db.query(key)

    # 3. 写入缓存
    if data:
        redis.set(key, data, ex=3600)

    return data

# 写流程
def update_data(key, value):
    # 1. 更新数据库
    db.update(key, value)

    # 2. 删除缓存（而非更新）
    redis.delete(key)

更新时删除缓存而非更新，避免并发写入导致脏数据。

Read-Through/Write-Through

# 缓存代理模式
# 应用只与缓存交互，缓存负责与数据库同步

class CacheProxy:
    def get(self, key):
        data = redis.get(key)
        if not data:
            data = db.query(key)
            redis.set(key, data)
        return data

    def set(self, key, value):
        db.update(key, value)
        redis.set(key, value)

Write-Behind（异步写入）

- 写入只更新缓存
- 异步批量写入数据库
- 提高写入性能
- 可能丢失数据（需权衡）

缓存更新策略

主动更新

# 数据更新时删除缓存
redis.delete(key)

# 下次读取时重新加载

定时更新

# 设置较短过期时间
redis.set(key, data, ex=300)

# 定时任务刷新缓存
# cron任务每5分钟刷新热点数据

被动更新

# 依赖过期自动刷新
# 读取时发现过期重新加载
redis.set(key, data, ex=3600)

缓存穿透

问题定义

查询不存在的数据：
- 缓存中没有
- 数据库中也没有
- 每次请求都打到数据库

场景：恶意攻击、爬虫

解决方案

方案1：缓存空值

def get_data(key):
    data = redis.get(key)
    if data == "NULL":  # 空值标记
        return None

    if data:
        return data

    data = db.query(key)
    if data:
        redis.set(key, data, ex=3600)
    else:
        # 缓存空值，防止穿透
        redis.set(key, "NULL", ex=60)  # 短过期

    return data

方案2：布隆过滤器

# 预热布隆过滤器
def preload():
    all_keys = db.query_all_keys()
    for key in all_keys:
        bloom_filter.add(key)

def get_data(key):
    # 布隆过滤器判断
    if not bloom_filter.might_contain(key):
        return None  # 确定不存在

    # 可能存在，继续查询
    data = redis.get(key)
    if data:
        return data

    data = db.query(key)
    if data:
        redis.set(key, data)

    return data

缓存击穿

问题定义

热点key过期：
- 高并发访问热点数据
- key刚好过期
- 大量请求同时查数据库

场景：热门商品、爆款文章

解决方案

方案1：互斥锁

def get_data(key):
    data = redis.get(key)
    if data:
        return data

    # 加锁
    lock_key = f"lock:{key}"
    if redis.set(lock_key, "1", nx=True, ex=10):
        # 获取锁成功
        try:
            data = db.query(key)
            redis.set(key, data, ex=3600)
        finally:
            redis.delete(lock_key)
        return data
    else:
        # 获取锁失败，等待重试
        sleep(0.1)
        return get_data(key)  # 重试

方案2：永不过期

# 热点数据不设置过期
redis.set(key, data)

# 后台定时更新
def refresh_cache():
    data = db.query(key)
    redis.set(key, data)

缓存雪崩

问题定义

大量key同时过期：
- 缓存集中过期
- 大量请求打到数据库
- 数据库压力骤增

场景：批量设置相同过期时间

解决方案

方案1：过期时间随机化

# 基础过期时间 + 随机偏移
base_expire = 3600
random_offset = random.randint(0, 300)
expire = base_expire + random_offset

redis.set(key, data, ex=expire)

方案2：多级缓存

架构：
L1缓存（本地缓存）→ L2缓存→ 数据库

- 本地缓存作为二级保护
- Redis缓存雪崩时本地缓存仍有数据

方案3：预热与监控

# 启动时预热热点数据
def warmup():
    hot_keys = get_hot_keys()
    for key in hot_keys:
        data = db.query(key)
        redis.set(key, data, ex=3600)

# 监控缓存命中率
cache_hit_rate = cache_hits / total_requests
if cache_hit_rate < 0.8:
    alert("缓存命中率过低")

缓存预热

启动预热

# 应用启动时加载热点数据
def init():
    # 从数据库加载热点数据
    hot_items = db.query_hot_items()
    for item in hot_items:
        redis.set(f"item:{item.id}", item.json(), ex=3600)

定时预热

# 定时刷新热点缓存
# 使用cron或调度任务
*/5 * * * * /path/to/refresh_cache.sh

缓存一致性

最终一致性

- 先更新数据库，再删除缓存
- 删除失败可重试
- 允许短暂不一致

强一致性（代价高）

- 使用分布式锁
- 读写串行化
- 性能下降明显

补偿机制

# 删除缓存失败时记录日志
def update_data(key, value):
    db.update(key, value)
    try:
        redis.delete(key)
    except:
        # 记录失败，定时重试
        log_failed_delete(key)

要点总结

• Cache-Aside最常用：先查缓存，不存在查数据库再写缓存
• 更新时删除缓存而非更新，避免并发脏数据
• 缓存穿透：缓存空值或使用布隆过滤器
• 缓存击穿：互斥锁或热点数据永不过期
• 缓存雪崩：过期时间随机化、多级缓存、预热
• 过期时间 = 基础时间 + 随机偏移，避免集中过期
• 热点数据预热，启动和定时加载
• 监控缓存命中率，低于阈值告警
• 缓存一致性：先更新数据库再删除缓存，允许短暂不一致