Redis缓存穿透击穿雪崩防护方案与热点Key治理实战

小编 • • 数据库

Redis缓存是提升系统吞吐量和降低数据库压力的核心手段，但缓存引入了新的故障模式：缓存穿透、缓存击穿和缓存雪崩。这三种异常场景如果缺乏防护，会导致大量请求穿透到数据库，引发数据库过载甚至服务雪崩。Redis缓存策略的设计不仅需要考虑正常场景的命中率，更要针对异常场景构建防护机制，配合热点Key治理确保缓存层的稳定性。

缓存异常场景分类与危害分析

缓存穿透：查询不存在的数据，缓存和数据库中都没有对应记录，每次请求都穿透缓存直达数据库。常见于恶意攻击（如查询不存在的ID）或业务异常请求。危害是数据库持续承受无效查询负载。

缓存击穿：热点Key在缓存中过期的瞬间，大量并发请求同时访问该Key，全部穿透到数据库。区别于雪崩，击穿是单个Key的并发问题。常见于秒杀商品的库存信息在过期时恰好被大量请求命中。

缓存雪崩：大量Key在同一时间集中过期，导致请求批量穿透到数据库。常见原因是缓存过期时间设置相同（如所有数据缓存1小时，1小时后全部过期）或Redis节点宕机导致缓存层整体不可用。

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

空值缓存是最简单的穿透防护方案。查询数据库未命中时，将空结果（null或特殊标记）写入缓存，设置较短的过期时间：

def get_user_with_null_cache(user_id):
    cache_key = f"user:{user_id}"
    cached = redis.get(cache_key)
    if cached is not None:
        if cached == "NULL_CACHE":
            return None
        return json.loads(cached)

    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        redis.setex(cache_key, 3600, json.dumps(user))
    else:
        redis.setex(cache_key, 300, "NULL_CACHE")
    return user

空值缓存的局限：如果攻击者使用大量不同的不存在ID进行请求，每个ID都会在缓存中占一个空值条目，内存消耗持续增长。布隆过滤器解决了这个问题。

布隆过滤器（Bloom Filter）是一种空间效率极高的概率型数据结构，用于判断元素是否在集合中。它的特点是：存在误判率（判断存在可能实际不存在），但不会漏判（判断不存在则一定不存在）。将所有合法ID加入布隆过滤器，请求先经过布隆过滤器过滤，不存在的ID直接返回：

from redisbloom.client import Client as RedisBloomClient

rb = RedisBloomClient(host='localhost', port=6379)

def init_bloom_filter():
    rb.bfCreate("user_ids", 0.001, 1000000)
    all_user_ids = db.query("SELECT id FROM users")
    for batch in chunk(all_user_ids, 1000):
        rb.bfMAdd("user_ids", *[uid for uid in batch])

def get_user_with_bloom(user_id):
    exists = rb.bfExists("user_ids", user_id)
    if not exists:
        return None

    cache_key = f"user:{user_id}"
    cached = redis.get(cache_key)
    if cached is not None:
        if cached == "NULL_CACHE":
            return None
        return json.loads(cached)

    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        redis.setex(cache_key, 3600, json.dumps(user))
    else:
        redis.setex(cache_key, 300, "NULL_CACHE")
    return user

布隆过滤器的关键参数：容量和误判率。容量需略大于实际数据量，误判率越低内存占用越大。100万数据量、0.1%误判率下，布隆过滤器仅需约1.4MB内存。新增数据需要同步加入布隆过滤器，删除数据时不能从布隆过滤器移除（不支持删除操作），可使用Counting Bloom Filter替代。

缓存击穿防护：热点Key互斥锁与逻辑过期

互斥锁方案在热点Key过期时，只允许一个请求查询数据库并重建缓存，其他请求等待或返回旧数据：

import time
import uuid

def get_hot_product_with_mutex(product_id):
    cache_key = f"product:hot:{product_id}"
    cached = redis.get(cache_key)
    if cached is not None:
        return json.loads(cached)

    lock_key = f"lock:product:{product_id}"
    lock_value = str(uuid.uuid4())
    acquired = redis.set(lock_key, lock_value, nx=True, ex=10)
    if acquired:
        try:
            # 双重检查
            cached = redis.get(cache_key)
            if cached is not None:
                return json.loads(cached)
            product = db.query("SELECT * FROM products WHERE id = %s", product_id)
            redis.setex(cache_key, 3600, json.dumps(product))
            return product
        finally:
            lua_release = '''
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            end
            '''
            redis.eval(lua_release, 1, lock_key, lock_value)
    else:
        time.sleep(0.05)
        cached = redis.get(cache_key)
        if cached is not None:
            return json.loads(cached)
        return None

互斥锁方案的要点是双重检查（DCL）：获取锁后再次检查缓存是否存在，避免重复查库。等待时间不宜过长，50-100ms足够缓存重建完成。锁过期时间设为10秒，略大于数据库查询时间，防止持锁线程异常导致锁无法释放。

逻辑过期方案不设置Redis TTL，在数据中携带逻辑过期时间，过期后异步重建缓存：

import json
import time
import threading

def get_product_logical_expire(product_id):
    cache_key = f"product:logical:{product_id}"
    cached = redis.get(cache_key)
    if cached is not None:
        data = json.loads(cached)
        if data.get("expire_time", 0) > time.time():
            return data["product"]
        # 已逻辑过期，异步重建
        lock_key = f"lock:rebuild:{product_id}"
        acquired = redis.set(lock_key, "1", nx=True, ex=30)
        if acquired:
            thread = threading.Thread(target=rebuild_cache, args=(product_id, cache_key), daemon=True)
            thread.start()
        return data["product"]
    # 冷启动，同步查库
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    cache_data = {"product": product, "expire_time": time.time() + 3600}
    redis.set(cache_key, json.dumps(cache_data))
    return product

def rebuild_cache(product_id, cache_key):
    try:
        product = db.query("SELECT * FROM products WHERE id = %s", product_id)
        cache_data = {"product": product, "expire_time": time.time() + 3600}
        redis.set(cache_key, json.dumps(cache_data))
    except Exception as e:
        log.error(f"缓存重建失败: {e}")
    finally:
        redis.delete(f"lock:rebuild:{product_id}")

缓存雪崩防护：随机过期时间与集群容灾

随机过期时间是最直接的雪崩防护措施。为每个Key的过期时间添加随机偏移量：

def set_cache_with_random_expire(key, value, base_expire=3600):
    random_offset = random.randint(-300, 300)
    expire_seconds = base_expire + random_offset
    redis.setex(key, expire_seconds, value)

def batch_set_cache(items, base_expire=3600):
    pipe = redis.pipeline()
    for key, value in items:
        random_expire = base_expire + random.randint(-600, 600)
        pipe.setex(key, random_expire, value)
    pipe.execute()

Redis集群容灾是雪崩防护的另一关键。容灾方案包括：Redis Sentinel哨兵模式实现自动故障转移；Redis Cluster分片集群单节点故障只影响部分Key；本地缓存兜底，Redis不可用时降级到本地缓存：

from cachetools import TTLCache

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

def get_with_fallback(key):
    try:
        value = redis.get(key)
        if value:
            local_cache[key] = value
            return json.loads(value)
    except redis.RedisError:
        if key in local_cache:
            return local_cache[key]
    value = db.query(key)
    if value:
        try:
            redis.setex(key, 3600, json.dumps(value))
        except redis.RedisError:
            pass
        local_cache[key] = json.dumps(value)
    return value

热点Key发现与治理方案

热点Key是指访问量远高于平均水平的缓存Key，如秒杀活动的商品Key、热门文章Key。热点Key会导致单节点Redis的CPU和网络带宽打满，即使集群模式下也会导致单个分片过载。

热点Key发现方法。Redis 4.0+提供hotkeys命令，需要在配置中启用LFU淘汰策略：

# redis.conf
maxmemory-policy allkeys-lfu
# 执行热点Key分析
redis-cli --hotkeys

监控层面，通过Redis的INFO commandstats和INFO latencystats分析各命令的执行频率和延迟。

热点Key治理方案。发现热点Key后，通过Key分散和本地缓存两个方向治理：

def get_hot_product_distributed(product_id):
    sub_keys = [f"product:hot:{product_id}:{i}" for i in range(10)]
    selected = random.choice(sub_keys)
    cached = redis.get(selected)
    if cached:
        return json.loads(cached)
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    pipe = redis.pipeline()
    for key in sub_keys:
        pipe.setex(key, 3600, json.dumps(product))
    pipe.execute()
    return product

Key分散方案将热点访问分散到多个Redis分片，降低单节点压力。读取时随机选择子Key，保证各子Key的访问概率均等。

缓存防护方案压测验证与监控指标

防护方案上线前需要压测验证。使用wrk或Locust模拟缓存过期瞬间的并发请求：

from locust import HttpUser, task, between

class CacheBreakdownUser(HttpUser):
    wait_time = between(0.01, 0.05)

    @task
    def access_hot_key(self):
        self.client.get(f"/api/product/hot/12345")

关键监控指标包括：缓存命中率（不低于95%为健康）、缓存穿透率（缓存未命中+数据库命中的比例）、数据库QPS（缓存生效时应远低于业务QPS）、Redis内存使用率和Key数量、热点Key访问频次分布。通过Prometheus采集RedisExporter指标和业务指标，设置告警：缓存命中率低于90%、数据库QPS突增超过阈值、Redis内存使用率超过80%时触发告警。

缓存预热是避免冷启动雪崩的有效手段。系统发布或重启后，缓存为空，大量请求直接打到数据库。预热方案是在系统启动时批量加载热点数据到缓存：

def warmup_cache():
    hot_products = db.query("""
        SELECT product_id, COUNT(*) as access_count
        FROM access_log
        WHERE created_at > NOW() - INTERVAL 7 DAY
        GROUP BY product_id
        ORDER BY access_count DESC
        LIMIT 1000
    """)
    pipe = redis.pipeline()
    for item in hot_products:
        product = db.query("SELECT * FROM products WHERE id = %s", item.product_id)
        expire = 3600 + random.randint(-300, 300)
        pipe.setex(f"product:{item.product_id}", expire, json.dumps(product))
    pipe.execute()
    log.info(f"缓存预热完成，加载 {len(hot_products)} 条数据")

预热过程应在系统接收流量前完成。对于全量预热耗时较长的场景，可以分批预热：先预热Top 100热点Key，开放流量后再异步预热剩余数据。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/redis-huan-cun-chuan-tou-ji-chuan-xue-beng-fang-hu-fang-an/

赞 (0)
小编小编
0
上一篇 12小时前
下一篇 12小时前

相关推荐