Redis缓存穿透与雪崩防护体系构建实战

缓存异常场景的分类与影响范围

Redis作为缓存层扛住了大部分读请求，但当缓存失效或异常时，请求直接打到数据库，可能引发级联故障。缓存异常分三种典型场景：缓存穿透（查询不存在的数据）、缓存击穿（热点Key过期瞬间）、缓存雪崩（大量Key同时过期）。三种场景的触发条件不同，防护方案也完全不同，混用方案不仅无效还可能引入新问题。

缓存穿透：布隆过滤器与空值缓存双保险

缓存穿透发生在查询数据库中不存在的数据时。恶意请求或爬虫大量访问不存在的ID，每次请求都穿透到数据库。

方案1：Redis布隆过滤器

在缓存层前置布隆过滤器，将所有合法数据ID映射到位数组，请求到达时先校验存在性：

import redis.commands.core.BloomFilterCommands

# Python redis-py 布隆过滤器使用
import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 初始化：加载所有合法用户ID到布隆过滤器
def init_bloom_filter(user_ids):
    pipe = r.pipeline()
    for uid in user_ids:
        pipe.execute_command('BF.ADD', 'user:bloom', uid)
    pipe.execute()

# 请求拦截
def get_user(user_id):
    # 布隆过滤器校验
    exists = r.execute_command('BF.EXISTS', 'user:bloom', user_id)
    if not exists:
        return None  # 数据一定不存在，直接拦截

    # 正常缓存查询流程
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 查库回填缓存
    user = db.query_user(user_id)
    if user:
        r.setex(cache_key, 3600, json.dumps(user))
    return user

布隆过滤器存在误判率（约0.1%-1%），不存在的数据有极小概率通过校验，但不影响正确性——漏网请求仍走正常缓存流程。误判率通过BF.RESERVE的error_rate参数控制，容量和误判率的权衡公式为m = -(n * ln(p)) / (ln2)^2。

方案2：空值缓存

对布隆过滤器不适用的大范围穿透（如分页查询任意页码），空值缓存是兜底方案：

def get_page_list(page, size):
    cache_key = f"page:{page}:{size}"
    cached = r.get(cache_key)

    if cached is not None:
        # 区分空值标记和正常数据
        if cached == "__NULL__":
            return []
        return json.loads(cached)

    # 查库
    data = db.query_page(page, size)

    if not data:
        # 空结果缓存60秒，防止穿透
        r.setex(cache_key, 60, "__NULL__")
    else:
        # 正常数据缓存5分钟
        r.setex(cache_key, 300, json.dumps(data))

    return data

空值缓存设置较短TTL（30-120秒），避免占用过多内存。穿透流量巨大时，可将空值TTL延长至5分钟，但需配合主动失效机制保证数据变更后缓存及时更新。

缓存击穿：热点Key的互斥锁重建

热点Key过期瞬间，数百个并发请求同时穿透到数据库重建缓存。解决方案是加互斥锁，只允许一个请求重建缓存，其余请求等待：

import time

def get_with_mutex(cache_key, fetch_fn, ttl=300, lock_timeout=10):
    # 互斥锁缓存重建
    data = r.get(cache_key)
    if data is not None:
        return json.loads(data)

    lock_key = f"lock:{cache_key}"

    # 尝试获取互斥锁
    acquired = r.set(lock_key, "1", nx=True, ex=lock_timeout)
    if acquired:
        try:
            # 获取锁成功，查库重建缓存
            data = fetch_fn()
            if data is not None:
                r.setex(cache_key, ttl, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    else:
        # 未获取锁，短暂等待后重试
        time.sleep(0.05)
        return get_with_mutex(cache_key, fetch_fn, ttl, lock_timeout)

SET NX EX实现分布式互斥锁，锁超时防止死锁。等待端50ms后重试，避免长时间阻塞。这套方案在峰值QPS 5000的热点Key场景下验证稳定。

缓存雪崩：过期时间偏移与多级缓存

雪崩的根因是大量Key在同一时刻集体过期。两个层面防护：

1. 过期时间添加随机偏移

import random

def set_with_jitter(cache_key, value, base_ttl=300, jitter=60):
    # 设置缓存时添加随机过期偏移
    actual_ttl = base_ttl + random.randint(0, jitter)
    r.setex(cache_key, actual_ttl, value)

# 批量设置时尤其重要
for item in items:
    ttl = 300 + random.randint(0, 120)  # 5分钟+0~2分钟随机偏移
    r.setex(f"item:{item.id}", ttl, json.dumps(item))

基础TTL 300秒，随机偏移0-120秒，使Key过期时间分散在5-7分钟窗口内，避免同时失效。

2. 本地缓存兜底

Redis整体不可用时，本地缓存作为最后防线：

from functools import lru_cache
import threading

# 二级缓存：本地 + Redis
local_cache = {}
local_cache_lock = threading.Lock()

def get_with_fallback(key, fetch_fn, redis_ttl=300, local_ttl=60):
    # L1: 本地缓存
    entry = local_cache.get(key)
    if entry and time.time() - entry['ts'] < local_ttl:
        return entry['data']

    # L2: Redis缓存
    try:
        data = r.get(key)
        if data is not None:
            data = json.loads(data)
            with local_cache_lock:
                local_cache[key] = {'data': data, 'ts': time.time()}
            return data
    except redis.RedisError:
        pass  # Redis异常，降级到本地缓存或查库

    # L3: 数据库
    data = fetch_fn()
    if data is not None:
        try:
            r.setex(key, redis_ttl, json.dumps(data))
        except redis.RedisError:
            pass
        with local_cache_lock:
            local_cache[key] = {'data': data, 'ts': time.time()}
    return data

Redis集群雪崩的熔断保护

当Redis集群整体故障时，应用层必须具备熔断能力，避免所有请求涌入数据库：

# 基于熔断器的Redis访问保护
class RedisCircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.last_failure_time = 0
        self.state = "closed"  # closed / open / half-open

    def call(self, fn, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
            else:
                raise CircuitBreakerOpen("Redis circuit breaker is open")

        try:
            result = fn(*args, **kwargs)
            if self.state == "half-open":
                self.state = "closed"
                self.failure_count = 0
            return result
        except redis.RedisError as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
            raise

缓存防护体系的核心思路是分层防御：布隆过滤器拦截无效请求、互斥锁防止击穿、随机过期防止雪崩、本地缓存兜底降级、熔断器保护数据库。每一层解决不同维度的风险，组合使用才能构建完整的缓存安全网。