Redis缓存策略设计与缓存穿透、击穿、雪崩防护实战方案

Redis缓存三大问题的本质区别

缓存穿透、缓存击穿、缓存雪崩是Redis运维中最常遇到的三类故障，本质都是缓存失效导致大量请求直达数据库：

• 穿透：请求查询根本不存在的数据，缓存和数据库都没有，每次请求都穿透到数据库
• 击穿：热点Key过期瞬间，大量并发请求同时打到数据库
• 雪崩：大量Key同时过期，或Redis节点宕机，请求洪峰压垮数据库

三种问题的防护策略各不相同，搞混了就治不了本。

缓存穿透防护：布隆过滤器与空值缓存

方案一：布隆过滤器前置拦截。所有可能存在的数据ID写入布隆过滤器，请求进来先过布隆过滤器判断：

# Redis布隆过滤器操作
# 添加元素
BF.ADD user_filter user_10001
BF.ADD user_filter user_10002

# 批量添加
BF.MADD user_filter user_10003 user_10004 user_10005

# 判断是否存在（不存在则一定不存在，存在可能误判）
BF.EXISTS user_filter user_10001  # 返回1
BF.EXISTS user_filter user_99999   # 返回0，拦截该请求

方案二：空值缓存。查询数据库不存在时，缓存一个空值并设置短过期时间：

def get_user(user_id):
    cache_key = f"user:{user_id}"
    cached = redis.get(cache_key)
    if cached is not None:
        if cached == "NULL":
            return None  # 命中空值缓存，不查库
        return json.loads(cached)

    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    if user is None:
        redis.setex(cache_key, 60, "NULL")  # 空值缓存60秒
        return None

    redis.setex(cache_key, 3600, json.dumps(user))
    return user

两种方案结合使用：布隆过滤器过滤明显非法的请求，空值缓存兜底处理误判漏过的少量穿透。

缓存击穿防护：互斥锁重建缓存

热点Key过期瞬间，只允许一个线程查库重建缓存，其他线程等待：

import redis
import time

redis_client = redis.Redis()

def get_hot_data(key):
    data = redis_client.get(key)
    if data is not None:
        return json.loads(data)

    lock_key = f"lock:{key}"
    lock_acquired = redis_client.set(lock_key, "1", nx=True, ex=10)

    if lock_acquired:
        try:
            data = db.query_hot_data(key)
            redis_client.setex(key, 3600, json.dumps(data))
            return data
        finally:
            redis_client.delete(lock_key)
    else:
        for _ in range(50):
            time.sleep(0.1)
            data = redis_client.get(key)
            if data is not None:
                return json.loads(data)
        raise Exception("缓存重建超时")

互斥锁方案保证同一时刻只有一个线程查库，其他线程等缓存重建完成后直接读缓存。适合热点Key少、重建耗时可控的场景。

缓存雪崩防护：过期时间随机化与多级缓存

方案一：给缓存Key的过期时间加随机偏移，避免同时失效：

import random

def set_cache_with_random_expire(key, value, base_expire=3600):
    expire = base_expire + random.randint(0, 300)
    redis_client.setex(key, expire, value)

方案二：本地缓存 + Redis二级缓存架构：

from cachetools import TTLCache

local_cache = TTLCache(maxsize=1000, ttl=300)

def get_data_with_l2_cache(key):
    if key in local_cache:
        return local_cache[key]

    data = redis_client.get(key)
    if data is not None:
        result = json.loads(data)
        local_cache[key] = result
        return result

    result = db.query_data(key)
    redis_client.setex(key, 3600, json.dumps(result))
    local_cache[key] = result
    return result

本地缓存挡住大部分读请求，即使Redis集群整体宕机，本地缓存仍能承担5分钟的缓冲窗口，给运维争取故障恢复时间。

缓存预热：避免冷启动瞬间压垮数据库

系统上线或大促前，主动加载热点数据到缓存：

def warm_up_cache():
    hot_items = db.query("""
        SELECT item_id, COUNT(*) as cnt
        FROM access_log
        WHERE created_at > DATE_SUB(NOW(), INTERVAL 7 DAY)
        GROUP BY item_id
        ORDER BY cnt DESC
        LIMIT 5000
    """)

    pipe = redis_client.pipeline()
    for item in hot_items:
        cache_key = f"item:{item['item_id']}"
        pipe.setex(cache_key, 7200, json.dumps(item))
    pipe.execute()
    print(f"预热完成，缓存加载数: {len(hot_items)}")

预热脚本在系统上线前15分钟执行，确保第一批用户请求到达时缓存已经就绪。

监控指标与告警阈值

缓存命中率是核心健康指标，低于90%需要告警。Redis的info命令输出keyspace_hits和keyspace_misses，计算命中率：hit_rate = keyspace_hits / (keyspace_hits + keyspace_misses)。配合Prometheus采集并设置告警规则，命中率持续5分钟低于85%触发P2告警。