Redis缓存策略实战：从缓存穿透到雪崩防护的完整防护体系

缓存失效的三种典型模式

Redis缓存的问题不是缓存本身，而是缓存失效后引发的一系列连锁反应。穿透、击穿、雪崩这三种模式本质上都是缓存失效后请求打到数据库的问题，区别在于失效的原因和波及范围不同。

缓存穿透防护方案

穿透的根源是恶意请求或无效查询绕过缓存直击数据库。防护分两层：

方案1：布隆过滤器前置拦截

在Redis前面加一层布隆过滤器，所有可能存在的key注册到布隆过滤器中，不存在的key直接被拦截：

# Python实现布隆过滤器缓存前置
import redis
import mmh3
import math

class RedisBloomFilter:
    def __init__(self, redis_client, key_prefix="bloom", capacity=1000000, error_rate=0.001):
        self.redis = redis_client
        self.key = key_prefix + ":" + str(capacity) + ":" + str(error_rate)
        self.bit_size = int(-capacity * (math.log(error_rate) / (math.log(2) ** 2)))
        self.hash_count = int(self.bit_size / capacity * math.log(2))

    def add(self, item):
        for i in range(self.hash_count):
            offset = mmh3.hash(str(item), i) % self.bit_size
            self.redis.setbit(self.key, offset, 1)

    def exists(self, item):
        for i in range(self.hash_count):
            offset = mmh3.hash(str(item), i) % self.bit_size
            if not self.redis.getbit(self.key, offset):
                return False
        return True

# 使用示例
bloom = RedisBloomFilter(r)
for uid in get_all_user_ids():
    bloom.add(uid)

def get_user(uid):
    if not bloom.exists(uid):
        return None
    return get_user_from_cache_or_db(uid)

方案2：空值缓存

对于穿透场景，即使数据库返回空也缓存一个短过期的空标记：

def get_with_null_cache(key, db_query, null_ttl=60, normal_ttl=3600):
    cached = redis_client.get(key)
    if cached is not None:
        if cached == b"__NULL__":
            return None
        return json.loads(cached)

    lock_key = "lock:" + key
    if redis_client.set(lock_key, "1", nx=True, ex=5):
        try:
            result = db_query()
            if result is None:
                redis_client.setex(key, null_ttl, "__NULL__")
            else:
                redis_client.setex(key, normal_ttl, json.dumps(result))
            return result
        finally:
            redis_client.delete(lock_key)
    else:
        time.sleep(0.05)
        return get_with_null_cache(key, db_query, null_ttl, normal_ttl)

缓存击穿防护方案

热点key过期的瞬间，大量并发请求同时发现缓存失效，全部涌向数据库。防护核心：只放一个请求去加载缓存，其余请求等待。

class HotKeyProtector:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.local_cache = {}
        self.local_lock = threading.Lock()

    def get(self, key, db_loader, lock_timeout=10, cache_ttl=3600):
        if key in self.local_cache:
            item = self.local_cache[key]
            if item["expire_at"] > time.time():
                return item["value"]

        cached = self.redis.get(key)
        if cached is not None:
            value = json.loads(cached)
            self._set_local(key, value, 30)
            return value

        lock_key = "hotlock:" + key
        if self.redis.set(lock_key, "1", nx=True, ex=lock_timeout):
            try:
                cached = self.redis.get(key)
                if cached is not None:
                    return json.loads(cached)
                value = db_loader()
                self.redis.setex(key, cache_ttl, json.dumps(value))
                self._set_local(key, value, 30)
                return value
            finally:
                self.redis.delete(lock_key)
        else:
            for _ in range(50):
                time.sleep(0.1)
                cached = self.redis.get(key)
                if cached is not None:
                    return json.loads(cached)
            raise Exception("缓存加载超时: " + key)

    def _set_local(self, key, value, ttl):
        with self.local_lock:
            self.local_cache[key] = {"value": value, "expire_at": time.time() + ttl}

缓存雪崩防护方案

雪崩的核心原因是大面积key同时过期。防护手段分三类：

1. 过期时间随机打散：

import random

def set_cache_with_jitter(key, value, base_ttl=3600, jitter_range=300):
    # 基础TTL + 随机偏移，避免同时过期
    actual_ttl = base_ttl + random.randint(-jitter_range, jitter_range)
    redis_client.setex(key, actual_ttl, json.dumps(value))

2. 熔断降级：

class CircuitBreaker:
    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"

    def allow_request(self):
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half_open"
                return True
            return False
        return True

    def record_success(self):
        self.failure_count = 0
        self.state = "closed"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"

3. 缓存预热：

def warmup_cache(hot_keys):
    for key_config in hot_keys:
        key = key_config["key"]
        loader = key_config["loader"]
        ttl = key_config["ttl"]
        try:
            value = loader()
            redis_client.setex(key, ttl, json.dumps(value))
        except Exception as e:
            print("预热失败 " + key + ": " + str(e))

warmup_cache([
    {"key": "hot:product:top100", "loader": lambda: db.get_top_products(100), "ttl": 1800},
    {"key": "hot:config:global", "loader": lambda: db.get_global_config(), "ttl": 3600},
])

缓存一致性保障

缓存和数据库不一致的问题无法100%消除，但可以把不一致窗口缩到最小：

延迟双删策略：

def update_with_cache(key, new_value, update_db_fn, delay=500):
    redis_client.delete(key)
    update_db_fn(new_value)
    threading.Timer(delay / 1000, lambda: redis_client.delete(key)).start()

基于Binlog的异步更新：订阅MySQL Binlog，数据变更时自动刷新缓存，实现秒级一致性。这是生产环境最可靠的方案，但需要Canal或Debezium等组件支撑。

缓存防护体系不是选一个方案就够的——穿透靠布隆过滤器+空值缓存，击穿靠分布式锁+本地缓存，雪崩靠TTL打散+熔断降级+预热，三层叠加才能扛住生产环境的复杂场景。

原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/redis-huan-cun-ce-lyue-shi-zhan-cong-huan-cun-chuan-tou-dao/