Redis 缓存穿透、击穿与雪崩应对

Redis 作为高性能的内存数据库，广泛应用于系统缓存层，有效减轻数据库压力并提升响应速度。然而，在高并发场景下，缓存系统可能面临穿透、击穿和雪崩等问题，这些问题会导致数据库压力剧增，甚至系统崩溃。本文将深入分析这三种缓存问题的原理，并提供相应的解决方案和代码示例。

缓存穿透

原理

缓存穿透是指查询一个根本不存在的数据，由于缓存中不存在，请求会直接打到数据库上。如果大量这类请求同时涌入，数据库将承受巨大压力。

应对策略

1. 缓存空对象：对于查询结果为空的数据，仍然将空结果缓存起来，并设置较短的过期时间
2. 布隆过滤器：使用布隆过滤器判断数据是否存在，不存在直接拦截

示例代码

# 缓存空对象示例
def get_user(user_id):
    # 先从缓存获取
    user = redis.get(f"user:{user_id}")
    
    if user is not None:
        if user == "NULL":  # 表示空对象
            return None
        return json.loads(user)
    
    # 缓存中没有，查询数据库
    user = db.query_user(user_id)
    
    if user:
        # 将结果存入缓存，设置过期时间
        redis.setex(f"user:{user_id}", 3600, json.dumps(user))
    else:
        # 缓存空对象，设置较短过期时间
        redis.setex(f"user:{user_id}", 60, "NULL")
    
    return user

# 使用布隆过滤器示例
from pybloom_live import ScalableBloomFilter

# 初始化布隆过滤器
bf = ScalableBloomFilter(initial_capacity=1000, error_rate=0.001)

# 添加已知ID到布隆过滤器
for user_id in get_all_user_ids():
    bf.add(user_id)

def get_user_with_bloom(user_id):
    # 先检查布隆过滤器
    if user_id not in bf:
        return None
    
    # 从缓存获取
    user = redis.get(f"user:{user_id}")
    
    if user is not None:
        if user == "NULL":
            return None
        return json.loads(user)
    
    # 缓存中没有，查询数据库
    user = db.query_user(user_id)
    
    if user:
        # 将结果存入缓存
        redis.setex(f"user:{user_id}", 3600, json.dumps(user))
        # 添加到布隆过滤器
        bf.add(user_id)
    else:
        # 缓存空对象
        redis.setex(f"user:{user_id}", 60, "NULL")
    
    return user

缓存击穿

原理

缓存击穿是指某个热点 key 在某一刻突然失效，大量并发请求直接打到数据库上，导致数据库压力骤增。

应对策略

1. 互斥锁：当缓存失效时，只允许一个请求查询数据库并重建缓存
2. 热点数据永不过期：对于热点数据，不设置过期时间，而是采用逻辑过期
3. 随机退避：对并发请求采用随机退避策略，避免同时重建缓存

示例代码

# 使用互斥锁防止缓存击穿
import threading

lock_dict = {}

def get_product_with_lock(product_id):
    # 先从缓存获取
    product = redis.get(f"product:{product_id}")
    
    if product is not None:
        return json.loads(product)
    
    # 获取锁
    lock = lock_dict.setdefault(f"product_lock:{product_id}", threading.Lock())
    acquired = lock.acquire(timeout=5)
    
    try:
        # 再次检查缓存，可能在等待锁时其他线程已经更新了缓存
        product = redis.get(f"product:{product_id}")
        if product is not None:
            return json.loads(product)
        
        # 缓存中没有，查询数据库
        product = db.query_product(product_id)
        
        if product:
            # 将结果存入缓存
            redis.setex(f"product:{product_id}", 3600, json.dumps(product))
        
        return product
    finally:
        if acquired:
            lock.release()
            # 可选：清理锁
            lock_dict.pop(f"product_lock:{product_id}", None)

# 逻辑过期示例
def get_product_with_logic_expire(product_id):
    # 先从缓存获取
    product_data = redis.get(f"product:{product_id}")
    
    if product_data is not None:
        product = json.loads(product_data)
        # 检查是否逻辑过期
        if product.get("expire_time", 0) > int(time.time()):
            return product
        # 已逻辑过期，标记为需要更新
        product["need_update"] = True
    
    # 获取锁
    lock = lock_dict.setdefault(f"product_lock:{product_id}", threading.Lock())
    acquired = lock.acquire(timeout=5)
    
    try:
        # 再次检查缓存
        product_data = redis.get(f"product:{product_id}")
        if product_data is not None:
            product = json.loads(product_data)
            if not product.get("need_update", False):
                return product
        
        # 缓存中没有或已逻辑过期，查询数据库
        product = db.query_product(product_id)
        
        if product:
            # 设置逻辑过期时间（例如1小时后）
            expire_time = int(time.time()) + 3600
            product["expire_time"] = expire_time
            # 将结果存入缓存
            redis.setex(f"product:{product_id}", 7200, json.dumps(product))
        
        return product
    finally:
        if acquired:
            lock.release()
            lock_dict.pop(f"product_lock:{product_id}", None)

缓存雪崩

原理

缓存雪崩是指大量 key 在同一时间失效，导致大量请求直接打到数据库上，造成数据库压力过大甚至宕机。

应对策略

1. 随机过期时间：为 key 设置基础过期时间加上一个随机值，避免同时失效
2. 多级缓存：使用本地缓存+分布式缓存的多级缓存架构
3. 服务降级与熔断：当检测到系统压力过大时，暂时降级或熔断部分服务
4. 高可用集群：使用 Redis 集群，避免单点故障

示例代码

import random
import time

# 随机过期时间
def set_product_with_random_expire(product_id, product):
    # 基础过期时间：1小时
    base_expire = 3600
    # 随机增加0-300秒的过期时间
    random_expire = random.randint(0, 300)
    total_expire = base_expire + random_expire
    
    # 将结果存入缓存
    redis.setex(f"product:{product_id}", total_expire, json.dumps(product))

# 多级缓存示例
from lru_cache import LRUCache

# 初始化本地缓存
local_cache = LRUCache(maxsize=1000)

def get_product_with_multi_cache(product_id):
    # 1. 先从本地缓存获取
    product = local_cache.get(product_id)
    if product is not None:
        return product
    
    # 2. 从 Redis 获取
    product_data = redis.get(f"product:{product_id}")
    if product_data is not None:
        product = json.loads(product_data)
        # 存入本地缓存
        local_cache.set(product_id, product)
        return product
    
    # 3. 缓存中没有，查询数据库
    product = db.query_product(product_id)
    
    if product:
        # 存入本地缓存
        local_cache.set