Redis 缓存策略与实战：解决缓存击穿、穿透、雪崩

小飞兽 Redis 4 次阅读 2026-07-29

前言

缓存是提升系统性能的关键技术，几乎所有高并发系统都离不开缓存。但引入缓存的同时，也带来了缓存击穿、穿透、雪崩这三大经典问题。本文不仅讲解问题的成因和解决方案，还提供完整的缓存架构设计思路。

缓存模式：Cache-Aside 与 Read-Through

在深入三大问题之前，先理解两种主流缓存模式：

Cache-Aside（旁路缓存）：应用先查缓存，缓存未命中再查数据库，然后写入缓存。读多写少场景首选。

Read-Through（穿透读）：应用只查缓存，缓存未命中时由缓存层自动查数据库并写入缓存，对应用透明。

# Cache-Aside 模式（最常用）
def get_user(user_id):
    cache_key = f"user:{user_id}"

    # Step 1: 查缓存
    cached = redis.get(cache_key)
    if cached:
        return json.loads(cached)

    # Step 2: 缓存未命中，查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", (user_id,))

    if user:
        # Step 3: 写入缓存，设置过期时间
        redis.setex(cache_key, 3600, json.dumps(user))

    return user


def update_user(user_id, data):
    # Step 1: 更新数据库
    db.execute("UPDATE users SET ... WHERE id = %s", (user_id, data))

    # Step 2: 删除缓存（删除而非更新，避免并发下的数据不一致）
    redis.delete(f"user:{user_id}")

缓存击穿（Cache Breakdown）

问题描述：一个热点 key 突然过期或不存在，大量并发请求直接打到数据库，导致数据库瞬时压力过大甚至崩溃。

解决方案：互斥锁（双检锁定）

def get_user_with_lock(user_id):
    cache_key = f"user:{user_id}"

    # 第一检：查缓存
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 获取分布式锁（SET NX EX 组合，原子操作）
    lock_key = f"lock:user:{user_id}"
    lock_acquired = r.set(lock_key, "1", nx=True, ex=10)

    if lock_acquired:
        try:
            # 获取锁成功后，再次检查缓存（第二检）
            cached = r.get(cache_key)
            if cached:
                return json.loads(cached)

            # 查数据库
            user = db.query("SELECT * FROM users WHERE id = %s", (user_id,))
            if user:
                r.setex(cache_key, 3600, json.dumps(user))
            return user
        finally:
            r.delete(lock_key)
    else:
        # 未获取到锁，短暂等待后重试
        import time
        for _ in range(50):
            time.sleep(0.1)
            cached = r.get(cache_key)
            if cached:
                return json.loads(cached)
        return db.query("SELECT * FROM users WHERE id = %s", (user_id,))

缓存穿透（Cache Penetration）

问题描述：查询一个根本不存在的数据（数据库中也没有），每次请求都穿透到数据库。如果被恶意利用，可能导致数据库被打垮。

解决方案1：缓存空值

def get_user_or_null(user_id):
    if user_id <= 0:
        return None

    cache_key = f"user:{user_id}"
    cached = r.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 = %s", (user_id,))

    if user:
        r.setex(cache_key, 3600, json.dumps(user))
    else:
        # 缓存空值，过期时间短一些（5 分钟）
        r.setex(cache_key, 300, "NULL")

    return user

解决方案2：Bloom Filter 布隆过滤器

# 使用 RedisBloom 模块或其他 Bloom Filter 实现
# 布隆过滤器判断 user_id 是否可能存在
# 可能存在（True）：再查缓存和数据库
# 一定不存在（False）：直接返回 None，不查数据库

# 初始化布隆过滤器（从数据库加载所有存在的 user_id）
def init_bloom_filter():
    all_user_ids = db.query("SELECT id FROM users")
    for uid in all_user_ids:
        bf.add(str(uid))

缓存雪崩（Cache Avalanche）

问题描述：大量缓存 key 在同一时间过期，导致大量请求同时穿透到数据库。不同于缓存击穿（单个热点 key），缓存雪崩是大量 key 的问题。

解决方案1：随机过期时间

def get_user_with_jitter(user_id):
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    user = db.query("SELECT * FROM users WHERE id = %s", (user_id,))
    if user:
        import random
        ttl = 3600 + random.randint(0, 720)  -- 基础 1 小时 + 0~12 分钟随机偏移
        r.setex(cache_key, ttl, json.dumps(user))

    return user

解决方案2：多级缓存架构

# L1（本地缓存）+ L2（Redis）+ L3（MySQL）多级缓存
import time

class MultiLevelCache:
    def __init__(self, redis_client):
        self.r = redis_client
        self.local_cache = {}
        self.local_expire = {}
        self.max_local = 10000

    def get(self, key):
        # L1: 查本地缓存
        if key in self.local_cache:
            if time.time() < self.local_expire[key]:
                return self.local_cache[key]
            del self.local_cache[key]
            del self.local_expire[key]

        # L2: 查 Redis
        cached = self.r.get(key)
        if cached:
            # 回填本地缓存
            if len(self.local_cache) >= self.max_local:
                oldest_key = min(self.local_expire, key=self.local_expire.get)
                del self.local_cache[oldest_key]
                del self.local_expire[oldest_key]
            self.local_cache[key] = cached
            self.local_expire[key] = time.time() + 60

        return cached

解决方案3：服务降级与熔断

def get_user_with_fallback(user_id):
    cache_key = f"user:{user_id}"

    try:
        cached = r.get(cache_key)
        if cached:
            return json.loads(cached)
    except RedisError:
        # Redis 故障，降级到数据库
        pass

    # 降级：直接查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", (user_id,))
    return user

缓存预热与更新策略

# 缓存预热：系统启动时将热点数据加载到缓存
def warmup_cache():
    hot_products = db.query("""
        SELECT * FROM products
        ORDER BY view_count DESC
        LIMIT 1000
    """)

    pipe = r.pipeline()
    for product in hot_products:
        cache_key = f"product:{product['id']}"
        import random
        pipe.setex(cache_key, 86400 + random.randint(0, 3600), json.dumps(product))
    pipe.execute()
    print(f"缓存预热完成，共加载 {len(hot_products)} 条数据")

常见问题

Q1：先更新数据库还是先删除缓存？
两种方案各有优缺。"先删缓存再更新数据库"可能造成短时间数据不一致，但通过延迟双删可以缓解。"先更新数据库再删除缓存"是更常见的选择，因为数据库是主数据源，缓存删除失败可以通过订阅 binlog 补偿。

Q2：如何确定缓存的过期时间？
根据数据变更频率和数据新鲜度要求决定。配置类数据可以设长一些（1-24 小时）；用户相关数据适中（15-60 分钟）；高频变更数据（库存、余额）建议不用 Redis 缓存。

Q3：Redis 内存满了会怎样？
如果 maxmemory 达到上限，Redis 会根据 maxmemory-policy 策略清理数据。常用策略包括：allkeys-lru（删除最近最少使用的 key）、volatile-lru（只删除设置了过期时间的 LRU key）、noeviction（拒绝写入）。生产环境建议设置 allkeys-lru 并监控内存使用率。