缓存穿透、击穿、雪崩解决方案

在缓存系统的使用过程中，三大经典问题——缓存穿透、缓存击穿和缓存雪崩，常常困扰着开发者。这些问题可能导致数据库压力骤增，甚至引发系统崩溃。本文将深入分析这三种问题的成因，并提供实用的解决方案。

一、缓存穿透

问题描述

缓存穿透是指查询一个根本不存在的数据，由于缓存中没有命中，每次查询都要到数据库中去查询，然后返回空。当并发量很大时，数据库可能承受不住压力而宕机。

典型场景：

• 恶意攻击，故意查询不存在的数据
• 业务逻辑缺陷，频繁查询已删除的数据

解决方案

方案1：缓存空对象

当数据库查询为空时，仍然将空结果缓存起来，设置较短的过期时间。

def get_user(user_id):
    # 先查缓存
    user = redis.get(f"user:{user_id}")
    if user is not None:
        return user if user != "NULL" else None
    
    # 缓存未命中，查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    
    if user:
        # 缓存实际数据，30分钟过期
        redis.setex(f"user:{user_id}", 1800, json.dumps(user))
    else:
        # 缓存空对象，5分钟过期
        redis.setex(f"user:{user_id}", 300, "NULL")
    
    return user
​

优点：实现简单
缺点：如果恶意攻击的key是随机的，会占用大量缓存内存

方案2：布隆过滤器（Bloom Filter）

布隆过滤器是一种空间效率极高的概率型数据结构，可以快速判断一个元素是否在一个集合中。

from pybloom_live import ScalableBloomFilter
import redis

# 初始化布隆过滤器
bloom = ScalableBloomFilter(initial_capacity=1000000, error_rate=0.001)
redis_client = redis.Redis()

def init_bloom_filter():
    """系统启动时，将所有有效key加载到布隆过滤器"""
    user_ids = db.query("SELECT id FROM users")
    for user_id in user_ids:
        bloom.add(user_id)

def get_user_v2(user_id):
    # 先用布隆过滤器判断
    if user_id not in bloom:
        # 确定不存在，直接返回
        return None
    
    # 可能存在，继续查缓存
    user = redis_client.get(f"user:{user_id}")
    if user:
        return json.loads(user)
    
    # 查数据库
    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        redis_client.setex(f"user:{user_id}", 1800, json.dumps(user))
    
    return user
​

优点：内存占用极小，适合海量key场景
缺点：存在误判率，需要容忍一定的false positive

二、缓存击穿

问题描述

缓存击穿是指某个热点key在缓存中过期的瞬间，大量并发请求同时查询这个key，导致所有请求都直接打到数据库上，可能瞬间压垮数据库。

典型场景：

• 秒杀商品信息缓存过期
• 热点新闻、微博热搜缓存过期

解决方案

方案1：互斥锁（Mutex Lock）

只允许一个线程查询数据库，其他线程等待。

import redis
import time

redis_client = redis.Redis()
lock_timeout = 10  # 锁超时时间，防止死锁

def get_product(product_id):
    # 查询缓存
    product = redis_client.get(f"product:{product_id}")
    if product:
        return json.loads(product)
    
    # 缓存未命中，尝试获取锁
    lock_key = f"lock:product:{product_id}"
    lock_value = str(time.time())
    
    # SETNX：只在key不存在时设置
    if redis_client.set(lock_key, lock_value, nx=True, ex=lock_timeout):
        try:
            # 获得锁，查询数据库
            product = db.query("SELECT * FROM products WHERE id = %s", product_id)
            
            # 双重检查：可能其他线程已经写入缓存
            if not redis_client.get(f"product:{product_id}"):
                redis_client.setex(f"product:{product_id}", 3600, json.dumps(product))
            
            return product
            
        finally:
            # 释放锁：使用Lua脚本确保只删除自己的锁
            lua_script = """
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
            """
            redis_client.eval(lua_script, 1, lock_key, lock_value)
    else:
        # 未获得锁，等待50ms后重试
        time.sleep(0.05)
        return get_product(product_id)  # 递归重试
​

优点：强一致性，保证只有一个线程查数据库
缺点：其他线程需要等待，可能影响响应时间

方案2：逻辑过期（Never Expire）

缓存数据不设置过期时间，而是在数据中增加逻辑过期时间字段。

def get_product_with_logical_expire(product_id):
    # 查询缓存
    cache_data = redis_client.get(f"product:{product_id}")
    
    if cache_data:
        data = json.loads(cache_data)
        expire_time = data.get('expire_time')
        
        # 检查逻辑过期时间
        if expire_time and time.time() < expire_time:
            # 未过期，直接返回
            return data['product']
        
        # 已过期，尝试获取锁更新缓存
        lock_key = f"lock:product:{product_id}"
        if redis_client.set(lock_key, "1", nx=True, ex=10):
            # 异步更新缓存
            threading.Thread(
                target=update_cache_async,
                args=(product_id,)
            ).start()
    
    # 如果缓存不存在或已过期，返回旧数据（如果有）
    return cache_data.get('product') if cache_data else None

def update_cache_async(product_id):
    """异步更新缓存"""
    try:
        product = db.query("SELECT * FROM products WHERE id = %s", product_id)
        cache_data = {
            'product': product,
            'expire_time': time.time() + 3600  # 逻辑过期时间
        }
        redis_client.set(f"product:{product_id}", json.dumps(cache_data))
    finally:
        redis_client.delete(f"lock:product:{product_id}")
​

优点：性能好，无需等待
缺点：可能返回过期数据，需要容忍一定的数据延迟

三、缓存雪崩

问题描述

缓存雪崩是指大量key在同一时间集中过期，或者缓存服务器故障宕机，导致所有请求都打到数据库上，造成数据库瞬时压力过大。

典型场景：

• 系统重启后，大量缓存同时失效
• 缓存服务器宕机
• 批量设置缓存时使用了相同的过期时间

解决方案

方案1：过期时间加随机值

在设置过期时间时，增加一个随机值，避免同时过期。

import random

def set_cache_with_random_expire(key, value, base_expire=3600):
    """设置缓存，过期时间为基础时间 + 随机时间"""
    random_expire = random.randint(1, 300)  # 0-5分钟的随机值
    total_expire = base_expire + random_expire
    
    redis_client.setex(key, total_expire, json.dumps(value))

# 批量设置缓存时使用
products = db.query("SELECT * FROM products")
for product in products:
    set_cache_with_random_expire(
        f"product:{product['id']}",
        product,
        base_expire=3600  # 基础1小时，实际会多0-5分钟
    )
​

方案2：缓存预热（Cache Warm-up）

系统启动时，提前加载热点数据到缓存中。

def cache_warmup():
    """缓存预热：系统启动时加载热点数据"""
    # 加载热点商品
    hot_products = db.query("""
        SELECT * FROM products 
        WHERE view_count > 10000 
        ORDER BY view_count DESC 
        LIMIT 1000
    """)
    
    for product in hot_products:
        set_cache_with_random_expire(
            f"product:{product['id']}",
            product,
            base_expire=3600
        )
    
    # 加载热点用户
    hot_users = db.query("""
        SELECT * FROM users 
        WHERE followers > 10000
        ORDER BY followers DESC
        LIMIT 1000
    """)
    
    for user in hot_users:
        set_cache_with_random_expire(
            f"user:{user['id']}",
            user,
            base_expire=1800
    )

# 在系统启动时调用
if __name__ == '__main__':
    cache_warmup()
​

方案3：多级缓存架构

使用本地缓存 + 分布式缓存的组合，即使分布式缓存挂了，本地缓存仍能服务部分请求。

from cachetools import TTLCache
import redis

# 本地缓存（LRU缓存，5分钟过期）
local_cache = TTLCache(maxsize=1000, ttl=300)
redis_client = redis.Redis()

def get_user_multi_level(user_id):
    # Level 1: 本地缓存
    if user_id in local_cache:
        return local_cache[user_id]
    
    # Level 2: Redis缓存
    user = redis_client.get(f"user:{user_id}")
    if user:
        user_data = json.loads(user)
        local_cache[user_id] = user_data  # 回填本地缓存
        return user_data
    
    # Level 3: 数据库
    user_data = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user_data:
        # 写入Redis缓存
        redis_client.setex(f"user:{user_id}", 1800, json.dumps(user_data))
        # 写入本地缓存
        local_cache[user_id] = user_data
    
    return user_data
​

方案4：熔断降级

当数据库压力过大时，自动降级，返回默认值或错误提示。

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def get_data_with_circuit_breaker(key):
    """带熔断保护的数据获取"""
    data = redis_client.get(key)
    if not data:
        data = db.query("SELECT * FROM table WHERE id = %s", key)
        if data:
            redis_client.setex(key, 3600, json.dumps(data))
    return data

# 使用时捕获异常
try:
    result = get_data_with_circuit_breaker(f"product:{product_id}")
except Exception:
    # 熔断触发，返回降级数据
    result = {"error": "服务繁忙，请稍后重试"}
​

最佳实践总结

通用建议：

1. 监控告警：对缓存命中率、数据库QPS、响应时间等指标进行监控
2. 限流保护：即使缓存失效，也要通过限流保护数据库
3. 定期演练：模拟缓存故障，验证系统的容错能力
4. 渐进式预热：避免系统启动时集中加载数据，分批次预热缓存

通过合理运用这些解决方案，可以构建一个高可用、高性能的缓存系统，有效应对各种异常场景。