Redis 深入——为什么你的缓存总是出问题 / Redis Internals and Cache Failure Modes

📅 创建时间：2026-05-08 🏷️ 标签：#Redis #缓存穿透 #缓存击穿 #缓存雪崩 #分布式锁 📚 相关知识：[[/03-web/06-databases-and-data-access/05-redis]]（Redis 数据结构基础） [[../Redis/01-redis-architecture]]（架构分析） [[../Redis/04-redis-distributed-locks]]（分布式锁原理） [[10-cache-strategy]]（缓存一致性） [[../Redis/02-redis-data-structures]]（数据结构场景）

---

场景：凌晨 2 点，你的系统被黑产打垮了

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  凌晨 2:15，你被报警电话叫醒。                             │
│                                                             │
│  监控大屏：                                                │
│  Redis 命中率：99% → 0%  ↓↓↓                            │
│  MySQL CPU：  5%  → 100%  ↑↑↑                            │
│  接口响应：  50ms → 超时                                 │
│                                                             │
│  错误日志清一色：                                          │
│  SELECT * FROM products WHERE id = 9999999                 │
│  SELECT * FROM products WHERE id = 8888888                 │
│  SELECT * FROM products WHERE id = 7777777                 │
│                                                             │
│  排查发现：有人用大量不存在的商品 ID，疯狂刷你的秒杀接口。  │
│                                                             │
└─────────────────────────────────────────────────────────────┘

这一章，我们从这个问题出发，一步步理解 Redis 缓存的三个核心问题：穿透、击穿、雪崩。

---

第1节：缓存穿透——恶意请求绕过缓存直达数据库

问题抽象

你开了一家图书馆，把热门书籍（热点数据）放在门口的展示柜（缓存）里，图书馆内部（数据库）是真正的书库。

正常情况：读者要书 → 先看展示柜 → 有就拿走（命中） → 没有就进图书馆找（未命中）→ 找到后放一本到展示柜

恶意情况：有人拿着一张根本不存在的书单来要书 → 展示柜没有 → 进图书馆找 → 图书馆也没有 → 不放入展示柜（因为没有这本书）→ 下次同样的请求再来 → 再次进图书馆找 → 死循环

推演：如果你是工程师

第一次请求 id=9999999：
用户请求 → 查 Redis（miss）→ 查 MySQL（无数据）→ 返回空 → Redis 不缓存空值

第二次请求 id=9999999：
用户请求 → 查 Redis（miss）→ 查 MySQL（无数据）→ 返回空
                    ↑
            每次都走数据库！
            10万个恶意ID = 10万次数据库查询 → 数据库爆炸

原理：什么是缓存穿透？

缓存穿透 = 查询一个根本不存在的数据（null），导致请求绕过缓存直接打到数据库。

恶意刷接口是最典型的场景，但正常业务中也可能发生（如商品下架后爬虫仍在请求）。

解法一：缓存空值（最简单）

def get_product(product_id):
    # 先查缓存
    cache_key = f"product:{product_id}"
    cache_val = redis.get(cache_key)

    if cache_val is not None:
        if cache_val == "NULL":  # 空值的标记
            return None
        return json.loads(cache_val)

    # 缓存 miss，查数据库
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)

    if product is None:
        # 空值缓存，过期时间短（5 分钟）
        redis.setex(cache_key, 300, "NULL")
    else:
        # 正常缓存，过期时间长
        redis.setex(cache_key, 3600, json.dumps(product))

    return product

适用场景：数据"确实不存在"的情况（如商品下架、用户注销）。注意过期时间不要太长，否则数据"出现"时会有不一致窗口。

解法二：布隆过滤器（最优雅）

┌─────────────────────────────────────────────────────────────┐
│                 布隆过滤器原理                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  布隆过滤器 = 一个位数组 + N 个哈希函数                    │
│                                                             │
│  存入 "商品ID=123"：                                       │
│  hash1(123) = 7   → 位数组[7] = 1                        │
│  hash2(123) = 15  → 位数组[15] = 1                       │
│  hash3(123) = 23  → 位数组[23] = 1                       │
│                                                             │
│  查询 "商品ID=123是否存在"：                              │
│  hash1(123) = 7   → 位数组[7] = 1 ✓                     │
│  hash2(123) = 15  → 位数组[15] = 1 ✓                     │
│  hash3(123) = 23  → 位数组[23] = 1 ✓                     │
│  → 三次都是 1 → 认为存在（可能误判，但不漏报）             │
│                                                             │
│  查询 "商品ID=9999999是否存在"：                          │
│  hash1(9999999) = 7   → 位数组[7] = 1 ✓                 │
│  hash2(9999999) = 2   → 位数组[2] = 0 ✗                 │
│  → 有 0 → 一定不存在！                                   │
│                                                             │
│  核心特性：                                               │
│  • 说"不存在" → 一定不存在（不会有漏报）                  │
│  • 说"存在" → 可能不存在（误判率可调，~1%）              │
│                                                             │
└─────────────────────────────────────────────────────────────┘

from bloom_filter import BloomFilter

# 启动时加载所有合法商品 ID 到布隆过滤器
bloom = BloomFilter(max_elements=10000000, error_rate=0.01)
for product_id in db.query("SELECT id FROM products"):
    bloom.add(str(product_id))

def get_product(product_id):
    # 第一关：布隆过滤器
    if str(product_id) not in bloom:
        return None  # 一定不存在，直接返回

    # 第二关：Redis 缓存
    cache_key = f"product:{product_id}"
    cache_val = redis.get(cache_key)
    if cache_val:
        return json.loads(cache_val)

    # 第三关：数据库
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    if product:
        redis.setex(cache_key, 3600, json.dumps(product))
    return product

适用场景：数据量大且相对固定（如全量商品 ID）。如果商品经常新增，需要定时重建布隆过滤器。

解法三：参数校验（最根本）

# 在网关层做最基础的校验
def check_request(product_id):
    # 负数 ID 一定不存在
    if product_id < 0:
        return False
    # ID 超出范围
    if product_id > MAX_PRODUCT_ID:
        return False
    return True

适用场景：永远不要相信用户输入。参数校验是最便宜也最有效的防线。

三种解法对比

┌─────────────────────────────────────────────────────────────┐
│              缓存穿透解法对比                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  │ 解法          │ 适用场景        │ 代价                │
│  ├───────────────┼────────────────┼─────────────────────┤
│  │ 缓存空值      │ 数据确实不存在  │ 内存浪费（存"NULL"）│
│  │ 布隆过滤器    │ 全量数据固定    │ 误判率 ~1%          │
│  │ 参数校验      │ 所有请求        │ 最便宜               │
│                                                             │
│  最佳实践：三种方案叠加使用                                 │
│  参数校验（第一道防线）→ 布隆过滤器（第二道）→ 缓存空值（兜底）│
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

第2节：缓存击穿——热点数据过期那一瞬间

问题抽象

还是图书馆的例子。一本超级热门的书（如《三体》）在展示柜里只有一本，大家都来借。

问题来了：这本书的"展示时间"（缓存过期）到了，要拿回图书馆重新放一本书进去。就在这个交接的瞬间，所有读者同时冲向图书馆——图书馆被挤爆了。

真实场景

┌─────────────────────────────────────────────────────────────┐
│                                                             │
│  凌晨 0 点，秒杀商品 A 的缓存刚刚过期。                    │
│                                                             │
│  10:00:00 - 100 个请求同时到达                             │
│  请求 1：查 Redis（过期了）→ 查 MySQL → 查到了            │
│  请求 2：查 Redis（过期了）→ 查 MySQL → 查到了            │
│  请求 3：查 Redis（过期了）→ 查 MySQL → 查到了            │
│  ...（100 个请求全部打到了数据库）                          │
│                                                             │
│  如果商品是爆款，可能 10 万个请求同时进来                    │
│  → 10 万次数据库查询 → 数据库崩溃                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

推演：如果你是工程师

第一反应：加互斥锁

请求 1 拿到了锁 → 查数据库 → 放入缓存 → 释放锁
请求 2 等待锁 → 锁被释放 → 查 Redis（有了！）→ 直接命中

这是对的，但...

互斥锁的问题：其他 99 个请求在等锁，响应时间变慢。
如果用的是 Redis 的 SETNX，性能尚可。
如果用的是数据库锁...那还不如一开始就用分布式锁。

更好的思路：热点数据永不过期

请求 1 拿到了锁 → 查数据库 → 放入缓存
但这次不设过期时间！
热点数据永远不会过期 → 永远不会击穿

但是...数据如何更新？
→ 用后台异步任务去刷新缓存
→ 用逻辑过期时间（value 里面存一个时间戳）

解法一：互斥锁（保守方案）

import redis
import time
import json

r = redis.Redis()

def get_product_with_lock(product_id):
    cache_key = f"product:{product_id}"

    # 先查缓存
    val = r.get(cache_key)
    if val:
        return json.loads(val)

    # 缓存 miss，获取互斥锁
    lock_key = f"lock:product:{product_id}"
    lock_id = str(uuid.uuid4())

    # SET NX EX：key 不存在才设置，并设置过期时间
    if r.set(lock_key, lock_id, nx=True, ex=10):
        try:
            # 拿到锁，查数据库
            product = db.query("SELECT * FROM products WHERE id = %s", product_id)
            if product:
                r.setex(cache_key, 3600, json.dumps(product))
            return product
        finally:
            # 释放锁（Lua 脚本保证原子性）
            lua = """
            if redis.call('get', KEYS[1]) == ARGV[1] then
                return redis.call('del', KEYS[1])
            else
                return 0
            end
            """
            r.eval(lua, 1, lock_key, lock_id)
    else:
        # 没拿到锁，等一下再试
        time.sleep(0.05)
        return get_product_with_lock(product_id)  # 递归重试

解法二：逻辑过期（推荐，热点数据）

def get_product_with_logic_expire(product_id):
    cache_key = f"product:{product_id}"

    val = r.get(cache_key)
    if not val:
        return get_and_cache_product(product_id)

    product_data = json.loads(val)
    expire_time = product_data["_expire_at"]
    now = time.time()

    # 逻辑过期：数据在，但时间到了
    if now > expire_time:
        # 开启一个独立线程去刷新缓存（不影响其他请求）
        thread = Thread(target=refresh_cache, args=(product_id,))
        thread.start()

        # 返回旧数据（短暂的不一致可以接受）
        return product_data

    return product_data

def get_and_cache_product(product_id):
    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    if not product:
        return None

    data = {
        **product,
        "_expire_at": time.time() + 300  # 逻辑过期时间，5 分钟
    }
    r.setex(cache_key, 86400, json.dumps(data))  # 物理过期时间设长
    return product

适用场景：热点数据（如商品详情页）。数据不一致的时间窗口很短（几十毫秒），用户体验几乎无感知。

解法三：热点数据永不过期（最激进）

# 启动时预热：将所有热点商品加载到 Redis，永远不设过期时间
def preload_hot_products():
    hot_products = db.query("SELECT * FROM products WHERE is_hot = 1")
    for product in hot_products:
        r.set(f"product:{product['id']}", json.dumps(product))
        # 不设置 EX，永不过期

# 后台任务：每 30 分钟刷新一次热点数据
def refresh_hot_products_periodically():
    while True:
        hot_products = db.query("SELECT * FROM products WHERE is_hot = 1")
        for product in hot_products:
            r.set(f"product:{product['id']}", json.dumps(product))
        time.sleep(1800)  # 30 分钟

适用场景：极端热点（如双十一的爆款）。缺点是数据更新需要靠主动推送，被动更新会有延迟。

---

第3节：缓存雪崩——同一时刻所有缓存同时失效

问题抽象

还是图书馆。展示柜里所有书的"展示时间"都是同一个时刻设置的。

灾难场景：凌晨 0 点，你设的"每日缓存过期时间"全部到期了。一瞬间，所有读者同时冲向图书馆——不是一本书，是所有书同时过期——图书馆被挤垮了。

真实场景

# 很多项目这样设缓存过期时间
def get_user(user_id):
    cache_key = f"user:{user_id}"
    if not redis.exists(cache_key):
        user = db.query("SELECT * FROM users WHERE id = %s", user_id)
        redis.setex(cache_key, 86400, json.dumps(user))  # 统一 24 小时过期
    return redis.get(cache_key)

# 问题：如果系统在 0 点大量用户注册/访问
# → 24 小时后凌晨 0 点，所有缓存同时过期
# → 所有请求同时穿透到数据库
# → 数据库瞬间压力暴增

推演：如果你是工程师

方案 1：随机过期时间
redis.setex(cache_key, 86400 + random.randint(0, 3600), ...)  # 24-25 小时随机

方案 2：不要用固定过期时间
不要对所有 key 设同样的过期时间

方案 3：多级缓存
L1（本地缓存，5 分钟过期）→ L2（Redis，30 分钟过期）→ MySQL
即使 Redis 全过期，本地缓存还能撑一下

方案 4：Redis 高可用
Redis Cluster + 哨兵，缓存节点不同时挂
但这个解决不了"同时 miss"的问题，只是加快恢复速度

解法：过期时间随机化 + 多级缓存

import random

def get_user(user_id):
    cache_key = f"user:{user_id}"
    val = redis.get(cache_key)
    if val:
        return json.loads(val)

    user = db.query("SELECT * FROM users WHERE id = %s", user_id)
    if user:
        # 过期时间加随机偏移量，避免同时失效
        expire = 86400 + random.randint(0, 3600)  # 24-25 小时随机
        redis.setex(cache_key, expire, json.dumps(user))
    return user

# 多级缓存架构
class MultiLevelCache:
    def __init__(self):
        self.l1_cache = {}  # 本地内存缓存（Guava/Caffeine）
        self.l1_ttl = 60    # L1: 60 秒

        self.redis = redis.Redis()
        self.l2_ttl = 1800  # L2: 30 分钟

    def get(self, key):
        # L1 优先
        if key in self.l1_cache:
            entry = self.l1_cache[key]
            if time.time() - entry["ts"] < self.l1_ttl:
                return entry["val"]
            del self.l1_cache[key]

        # L2 其次
        val = self.redis.get(key)
        if val:
            # 回填 L1
            self.l1_cache[key] = {"val": json.loads(val), "ts": time.time()}
            return json.loads(val)

        return None

---

第4节：Redis 分布式锁——秒杀系统的核心

问题抽象

回到秒杀场景。10 万人同时抢 100 部手机。怎么保证这 100 部手机不会"超卖"？

方案 A：数据库加锁

BEGIN;
SELECT stock FROM products WHERE id = 1;  -- stock = 100
UPDATE products SET stock = stock - 1 WHERE id = 1 AND stock > 0;
COMMIT;

问题：10 万 QPS 同时打 MySQL → 事务排队 → 数据库崩溃

方案 B：Redis 分布式锁

抢手机：
SET lock:product:1 unique_id NX EX 30
→ 抢到锁的人去买手机 → 买完释放锁
→ 没抢到锁的人...直接返回"已售罄"（不用进数据库）

推演：分布式锁的正确实现

# 错误实现：先 SET 再 EXPIRE（两步骤，有竞态）
r.set("lock", "123")    # 步骤1：设值
r.expire("lock", 30)    # 步骤2：设过期
# 问题：如果步骤1和2之间进程崩溃了 → 锁永远不过期 → 死锁

# 正确实现：SET NX EX（单指令，原子性）
r.set("lock:product:1", lock_id, nx=True, ex=30)

# 错误释放：用 DEL 删除锁（可能删别人的锁）
r.delete("lock:product:1")
# 问题：锁 A 过期了，锁 A 自动释放；锁 B 拿到了锁；锁 A 才想起来 DEL → 删掉了锁 B

# 正确释放：Lua 脚本，只删除自己的锁
lua = """
if redis.call('get', KEYS[1]) == ARGV[1] then
    return redis.call('del', KEYS[1])
else
    return 0
end
"""
r.eval(lua, 1, lock_key, lock_id)

分布式锁完整实现

import redis
import uuid
import time
import threading

class RedisLock:
    def __init__(self, redis_client, key, timeout=30):
        self.redis = redis_client
        self.key = f"lock:{key}"
        self.timeout = timeout
        self.lock_id = str(uuid.uuid4())
        self.acquired = False

    def acquire(self, blocking=True, blocking_timeout=10):
        end = time.time() + blocking_timeout
        while time.time() < end:
            # SET NX EX：原子获取锁
            if self.redis.set(self.key, self.lock_id, nx=True, ex=self.timeout):
                self.acquired = True
                return True

            if not blocking:
                return False
            time.sleep(0.01)  # 10ms 重试

        return False

    def release(self):
        """只删除自己持有的锁"""
        if not self.acquired:
            return

        lua = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('del', KEYS[1])
        else
            return 0
        end
        """
        self.redis.eval(lua, 1, self.key, self.lock_id)
        self.acquired = False

    def extend(self, add_seconds=10):
        """锁续期（自动续命，防止处理时间过长导致锁自动释放）"""
        lua = """
        if redis.call('get', KEYS[1]) == ARGV[1] then
            return redis.call('expire', KEYS[1], ARGV[2])
        else
            return 0
        end
        """
        self.redis.eval(lua, 1, self.key, self.lock_id, self.timeout + add_seconds)

# Redisson 的 watchdog 机制：自动续期
# 如果持有锁的进程还没执行完，watchdog 每 10 秒自动续期
# 只有主动释放锁或进程崩溃时才会真正释放

可重入锁（同一个进程可以多次获取）

# 普通锁：同一进程两次 acquire 会死锁
# 可重入锁：同一进程可以多次获取，只有释放次数等于获取次数时才真正释放

lua_acquire = """
local key = KEYS[1]
local id = ARGV[1]
local ttl = ARGV[2]

-- 如果锁不存在，初始化
if redis.call('exists', key) == 0 then
    redis.call('hset', key, id, 1)
    redis.call('expire', key, ttl)
    return 1
end

-- 如果是自己持有的锁，重入计数 +1
if redis.call('hget', key, id) then
    redis.call('hincrby', key, id, 1)
    redis.call('expire', key, ttl)
    return 1
end

return 0
"""

---

升华：穿透/击穿/雪崩的串联理解

┌─────────────────────────────────────────────────────────────┐
│              三个问题的本质区别                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  穿透：数据根本不存在                                       │
│  → 查缓存（miss）→ 查数据库（也没有）→ 不缓存空值          │
│  → 解决方案：缓存空值 / 布隆过滤器 / 参数校验              │
│                                                             │
│  击穿：热点数据过期了（大量请求同时进来）                   │
│  → 查缓存（miss）→ 查数据库（同时 10 万次）               │
│  → 解决方案：互斥锁 / 逻辑过期 / 热点永不过期            │
│                                                             │
│  雪崩：所有缓存同时过期                                     │
│  → 查缓存（大量 miss）→ 查数据库（同时爆炸）              │
│  → 解决方案：过期时间随机化 / 多级缓存 / Redis 高可用     │
│                                                             │
│  共同点：大量请求穿透到数据库，导致数据库崩溃               │
│  区别：触发原因不同，解决方案不同                         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

---

"AI 可查 vs 必须理解"清单

AI 可查：
✅ Redis SET/SETNX/EXPIRE/Lua 脚本的具体命令语法
✅ 布隆过滤器的参数计算公式（n, p → m, k）
✅ Redisson 的 API 用法

必须理解：
🔴 穿透/击穿/雪崩三个问题的本质区别
🔴 互斥锁的 Lua 脚本为什么能保证原子性
🔴 为什么 SET NX EX 比 SET + EXPIRE 更安全
🔴 逻辑过期的"短暂不一致"是什么量级
🔴 分布式锁的四个核心问题：获取/释放/续期/可重入

---

学习状态：🟡 开始学习