如何解决缓存击穿？

解决缓存击穿问题的核心思路是防止热点key失效时大量请求直接访问数据库。以下是几种常见的解决方案及其实现要点：

1. 永不过期策略（逻辑过期）

原理：

• 缓存key不设置物理过期时间（TTL），而是在value中存储一个逻辑过期时间戳。
• 访问时判断逻辑时间是否过期，若过期则异步更新缓存，同时返回旧数据。

优点：

• 完全避免缓存失效瞬间的穿透问题。
• 适用于对数据实时性要求不高的场景（如商品详情、配置信息）。

实现示例：

def get_data(key):
    data = redis.get(key)
    if data:
        # 检查逻辑过期时间
        if data["expire_time"] < time.time():
            # 异步更新缓存（如通过线程池或消息队列）
            thread_pool.submit(update_cache, key)
        return data["value"]
    else:
        # 缓存中没有数据（罕见情况），查库并更新
        return update_cache(key)

def update_cache(key):
    # 加锁防止并发更新
    with redis.lock(f"lock:{key}"):
        # 再次检查缓存是否已被其他线程更新
        if redis.get(key)["expire_time"] > time.time():
            return redis.get(key)["value"]
        # 查询数据库
        value = db.query(key)
        # 设置新值和逻辑过期时间（如30分钟后）
        redis.set(key, {
   
            "value": value,
            "expire_time": time.time() + 1800
        })
        return value

2. 互斥锁（分布式锁）

原理：

• 当缓存失效时，仅允许一个请求（通过获取锁）访问数据库，其他请求等待锁释放后从缓存获取数据。

优点：

• 保证数据库仅被一个请求访问，压力可控。
• 实现简单，兼容性强。

缺点：

• 锁竞争可能导致部分请求等待，影响吞吐量。

实现示例：

def get_data(key):
    data = redis.get(key)
    if data:
        return data

    # 尝试获取锁
    if redis.setnx(f"lock:{key}", "1", ex=10):  # 锁超时10秒
        try:
            # 查询数据库
            data = db.query(key)
            # 更新缓存（设置合理TTL）
            redis.set(key, data, ex=3600)
            return data
        finally:
            # 释放锁
            redis.delete(f"lock:{key}")
    else:
        # 未获取到锁，等待重试
        time.sleep(0.1)  # 短暂休眠避免频繁重试
        return get_data(key)  # 递归重试

3. 热点数据预热

原理：

• 通过定时任务或监控系统，在缓存过期前主动刷新热点key，确保缓存始终有效。

适用场景：

• 已知的热点数据（如首页banner、热门榜单），且有明确访问模式。

实现示例：

# 定时任务（如每20分钟执行一次）
def refresh_hot_keys():
    hot_keys = ["product:1001", "product:1002", "category:hot"]  # 热点key列表
    for key in hot_keys:
        # 查询数据库
        data = db.query(key)
        # 更新缓存（设置TTL比刷新周期长）
        redis.set(key, data, ex=30*60)  # 30分钟过期，任务每20分钟执行一次

4. 熔断降级

原理：

• 当检测到数据库压力过大（如响应超时、错误率上升）时，临时返回默认值或旧数据，避免请求穿透到数据库。

工具：

• 可使用Sentinel、Hystrix等熔断框架实现。

实现示例：

from circuitbreaker import CircuitBreaker

# 配置断路器：连续3次失败则开启熔断，5秒后尝试恢复
breaker = CircuitBreaker(fail_max=3, reset_timeout=5)

@breaker
def query_db(key):
    return db.query(key)  # 可能抛出异常的数据库查询

def get_data(key):
    data = redis.get(key)
    if data:
        return data

    try:
        # 尝试查询数据库（可能触发熔断）
        data = query_db(key)
        redis.set(key, data, ex=3600)
        return data
    except CircuitBreakerError:
        # 熔断开启，返回默认值或旧数据
        return {
   "message": "Service temporarily unavailable", "data": None}

5. 二级缓存（LocalCache + Redis）

原理：

• 在分布式缓存（如Redis）之上增加本地缓存（如Guava Cache、Caffeine），热点数据优先从本地读取，减少对Redis的访问。

优点：

• 降低网络延迟，进一步减轻Redis压力。

实现示例：

import caffeine

# 初始化本地缓存（最大1000条，10分钟过期）
local_cache = caffeine.CacheBuilder.newBuilder() \
    .maximumSize(1000) \
    .expireAfterWrite(10, TimeUnit.MINUTES) \
    .build()

def get_data(key):
    # 优先从本地缓存获取
    data = local_cache.get_if_present(key)
    if data:
        return data

    # 从Redis获取
    data = redis.get(key)
    if data:
        # 放入本地缓存
        local_cache.put(key, data)
        return data

    # 缓存失效，查库并更新
    data = db.query(key)
    redis.set(key, data, ex=3600)
    local_cache.put(key, data)
    return data

方案选择建议

实际应用中，通常会组合多种方案（如热点预热 + 互斥锁 + 熔断降级），以应对不同场景下的缓存击穿风险。