当前位置：首页 > news >正文

Redis缓存雪崩、击穿、穿透

news 2025/11/5 2:24:27

缓存雪崩、缓存击穿、缓存穿透
这三个问题都发生在缓存失效或不存在时，大量请求直接涌向后端数据库，导致数据库压力激增甚至崩溃。

一、缓存雪崩 (Cache Avalanche)

1. 问题描述

同一时间，大量的缓存Key集体失效（例如，设置了相同的过期时间），导致所有对这些数据的请求同时无法命中缓存（Cache Miss），全部直接落到数据库上，引起数据库瞬时压力过大而崩溃。

2. 发生场景

业务高峰期，一批缓存数据（如首页商品列表、热门文章等）设置了相同的 TTL（Time-To-Live），同时过期。
Redis 服务宕机或重启，导致整个缓存层不可用。

3. 解决方案

解决方案的核心是：避免大量Key同时失效。

设置随机过期时间
这是最简单有效的预防措施。给缓存数据的过期时间加上一个随机值，打散它们的失效时间点，避免集体失效。

import random
# 设置缓存时，在基础过期时间上增加一个随机抖动（例如0-300秒）
expire_time = 3600 + random.randint(0, 300) # 1小时 ± 5分钟
redis_client.setex(cache_key, expire_time, data)

构建高可用的Redis集群
通过 Redis Sentinel（哨兵）或 Redis Cluster（集群）模式，实现主从切换和多节点负载均衡，即使单个节点宕机，整个缓存层依然可用，防止“全盘皆崩”。
缓存永不过期 + 后台更新
对极热点数据，可以设置为永不过期（-1），然后由后台任务或定时任务定期异步地更新缓存。这样用户请求永远不会遇到缓存失效。

二、缓存击穿 (Cache Breakdown)

1. 问题描述

一个访问量极高的热点Key（如某顶流明星的新闻、秒杀商品）在失效的瞬间，持续的高并发请求会像子弹一样“击穿”缓存，全部直接请求数据库，导致数据库瞬间压力过大。

2. 解决方案

解决方案的核心是：防止单个热点Key失效时被大量并发访问。

互斥锁 (Mutex Lock) - 最常用
当缓存失效时，不是所有请求都去查数据库，而是让第一个请求去查数据库并重建缓存，其他请求等待，待缓存重建后再从缓存中获取数据。

import redis
import threading
from datetime import datetimedef get_data_with_lock(key):# 1. 先尝试从缓存获取data = redis_client.get(key)if data is not None:return data# 2. 缓存未命中，尝试获取分布式锁lock_key = f"lock:{key}"# 使用 setnx (SET if Not eXists) 命令争抢锁，并设置锁的过期时间（防止死锁）acquired_lock = redis_client.setnx(lock_key, datetime.now().strftime("%s"))if acquired_lock:redis_client.expire(lock_key, 10)  # 设置锁10秒后自动过期try:# 3. 成功获取锁，查询数据库data = get_data_from_db(key)# 4. 写入缓存redis_client.setex(key, 3600, data)finally:# 5. 释放锁redis_client.delete(lock_key)return dataelse:# 6. 未获取到锁，等待片刻后重试（或直接返回默认值）time.sleep(0.1)return get_data_with_lock(key)  # 重试

逻辑过期 (Logical Expiration)
不给缓存设置 TTL，而是在缓存Value中存储一个逻辑过期时间。当请求发现逻辑时间已过期，则发起一个异步任务去更新缓存，当前请求仍返回旧数据。

# Value 结构：{"data": real_data, "expire_ts": 1649873100}
value = {"data": {"name": "Hot Product", "price": 99},"expire_ts": int(time.time()) + 3600  # 1小时后逻辑过期
}
redis_client.set(key, json.dumps(value)) # 不设置TTL# 读取时：
cached_value = redis_client.get(key)
if cached_value:obj = json.loads(cached_value)if obj['expire_ts'] > time.time():return obj['data']  # 未逻辑过期，直接返回else:# 已逻辑过期，触发异步更新async_update_cache(key)return obj['data']  # 仍然先返回旧数据

三、缓存穿透 (Cache Penetration)

1. 问题描述

请求查询一个数据库中根本不存在的数据（如 id=-1 的商品，或随机生成的、不存在的用户ID）。由于缓存中也不会有该数据，导致每次请求都会穿透缓存去查询数据库。如果有人恶意攻击，会发送大量此类请求，从而压垮数据库。

2. 解决方案

解决方案的核心是：在缓存层拦截掉对不存在数据的请求。

缓存空对象 (Cache Null Object)
即使从数据库没查到，也缓存一个空值（如 None, NULL）或特定的错误标记，并设置一个较短的过期时间（如 1-5 分钟）。后续相同的请求会命中这个空缓存，从而保护数据库。

def get_data(key):data = redis_client.get(key)if data is not None:# 如果缓存的是空标记，直接返回None或错误if data == "NULL_OBJECT":return Nonereturn data# 查数据库data = db.query("SELECT * FROM table WHERE id = %s", key)if not data:# 数据库不存在，缓存空对象，有效期5分钟redis_client.setex(key, 300, "NULL_OBJECT")return Noneelse:# 数据库存在，写入缓存redis_client.setex(key, 3600, data)return data

缺点：如果攻击者每次用不同的Key，此方法会缓存大量无用的空值，浪费内存。

布隆过滤器 (Bloom Filter) - 最优解
在缓存之前，设置一个布隆过滤器。它是一个概率型数据结构，用于快速判断一个元素是否绝对不存在于某个集合中。

写入时：当向数据库插入新数据时，同时将该数据的Key写入布隆过滤器。
查询时：收到请求后，先用布隆过滤器判断Key是否存在。
- 如果不存在，则直接返回 None，根本不会查询缓存和数据库。
- 如果存在，再继续后续的缓存查询流程。
```
# 使用 Redis 4.0+ 自带的布隆过滤器模块 (redisbloom)
# 或者使用 Python 的 redisbloomclient 库from redisbloom.client import Clientrb = Client()
key = "user:10000"# 1. 先检查布隆过滤器
if not rb.bfExists("users_filter", key):print("Key definitely not exists, return directly.")return None# 2. 如果布隆过滤器说存在，再继续查缓存和数据库
data = redis_client.get(key)
if data:return data
# ... (后续流程)
```
优点：内存占用极小，效率极高。
缺点：有极低的误判率（判断为存在，但实际可能不存在），但不会误判“不存在”。对于缓存穿透场景，宁可错杀一千（放过极少数不存在的请求去查库），也不能放过一个。