原子性、分布式锁与缓存经典问题深度解析

面试开场白

面试官：Redis 的原子性是怎么保证的？分布式锁怎么实现？

你：Redis 单命令天然原子，因为单线程执行。复杂操作用 Lua 脚本保证原子性。分布式锁用 SET key value NX EX 秒数，解锁用 Lua 脚本先验证 value 再删除，防止误删其他线程的锁。

面试官：那如果锁过期了但业务还没执行完怎么办？Redisson 的看门狗机制了解吗？

这个追问考察是否真正用过分布式锁。能说出看门狗续期机制的，证明有实际生产经验。

链式追问一：Redis 的原子性

Q1：Redis 如何保证原子性？MULTI/EXEC 事务和数据库事务有什么区别？必考

回答要点：

Redis 的三种原子性级别：

级别	实现方式	原子性保证	适用场景
单命令	单线程执行	✅ 天然原子	INCR、SETNX、GETSET
事务	MULTI/EXEC	⚠️ 不支持回滚	批量执行，不依赖中间结果
Lua 脚本	EVAL/EVALSHA	✅ 整个脚本原子	CAS 操作、复杂逻辑

MULTI/EXEC 事务的问题：

MULTI
SET k1 v1
INCR k2        -- k2 不是整数，会失败
SET k3 v3
EXEC

-- 结果：
-- k1 设置成功
-- k2 失败（ERR value is not an integer or out of range）
-- k3 仍然设置成功！不会回滚 ❌

对比 MySQL 事务：

维度	MySQL 事务	Redis MULTI/EXEC
原子性	✅ 要么全成功，要么全回滚	⚠️ 运行时错误不回滚
隔离性	✅ MVCC/锁，多种隔离级别	✅ 单线程，EXEC 期间不处理其他请求
持久性	✅ redo log	⚠️ 依赖持久化策略（RDB/AOF）
一致性	✅ 外键、触发器、约束	❌ 无约束

为什么 Redis 事务不支持回滚？

Antirez（Redis 作者）的理由：
1. Redis 命令错误只有两种：
   - 语法错误（如参数不对）：整个事务不执行
   - 类型错误（如对 String 执行 LPUSH）：只有该命令失败
2. 类型错误是编程错误，应该在开发阶段发现，不应该在生产环境发生
3. 支持回滚需要复杂的事务日志，增加内存开销和实现复杂度
4. Redis 追求简单高效，不支持回滚是权衡的结果

Lua 脚本的原子性：

-- CAS（Check-And-Set）操作
local val = redis.call('GET', KEYS[1])
if val == ARGV[1] then
    return redis.call('SET', KEYS[1], ARGV[2])
end
return 0

执行 Lua 脚本时：
  1. Redis 主线程独占执行整个脚本
  2. 期间不处理其他命令（完全原子）
  3. 脚本执行完，才会处理后续命令

本质一句话：Redis 单命令和 Lua 脚本天然原子，MULTI/EXEC 不支持回滚，只保证 EXEC 期间的隔离性。

Q2：Lua 脚本在 Redis 中有什么优势？有什么注意事项？高频

回答要点：

Lua 脚本的优势：

1. 原子性：整个脚本原子执行，不会被其他命令插入
2. 减少网络开销：多个命令一次发送，减少网络往返
3. 复用性：EVALSHA 执行缓存的脚本，节省带宽
4. 灵活性：实现复杂逻辑，如 CAS、限流、分布式锁

代码示例：

-- 示例1：限流脚本（令牌桶）
local tokens = redis.call('GET', KEYS[1])
if not tokens then
    tokens = ARGV[1]  -- 初始令牌数
end
if tonumber(tokens) > 0 then
    redis.call('DECR', KEYS[1])
    redis.call('EXPIRE', KEYS[1], ARGV[2])
    return 1  -- 允许访问
else
    return 0  -- 拒绝访问
end

-- 调用方式
EVAL script 1 rate_limit:user:1001 100 3600
-- KEYS[1] = rate_limit:user:1001
-- ARGV[1] = 100（令牌数）
-- ARGV[2] = 3600（过期时间）

注意事项：

问题1：脚本执行时间长会阻塞主线程
  解决：避免复杂计算、死循环，脚本执行时间 < 5ms

问题2：脚本缓存占用内存
  解决：定期 SCRIPT FLUSH 清理无用脚本

问题3：脚本错误会中断执行
  解决：用 redis.pcall 代替 redis.call（捕获错误）

性能对比：

测试场景：执行 100 次 GET + SET（无依赖）
- 普通方式：100 次 GET + 100 次 SET = 200 次网络往返，耗时约 200ms
- Pipeline：1 次网络往返（批量发送），耗时约 10ms
- Lua 脚本：1 次网络往返 + 原子执行，耗时约 8ms

结论：Lua 脚本和 Pipeline 都能减少网络开销，但 Lua 提供原子性保证。

本质一句话：Lua 脚本提供原子性、减少网络开销、支持复杂逻辑，但需避免长时间阻塞主线程。

链式追问二：分布式锁实现

Q3：Redis 分布式锁的正确实现是什么？为什么 value 要用唯一值？必考

回答要点：

错误实现（旧版）：

SETNX lock:order:1001 1     -- 加锁
EXPIRE lock:order:1001 30   -- 设置过期时间

问题：
  1. SETNX 和 EXPIRE 不是原子的
  2. SETNX 成功后进程崩溃，EXPIRE 未执行 → 死锁

正确实现（Redis 2.6.12+）：

-- 加锁：SET key value NX EX seconds（原子命令）
SET lock:order:1001 "uuid-12345" NX EX 30

-- 解锁：用 Lua 脚本保证原子性
EVAL "
  if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
  else
    return 0
  end
" 1 lock:order:1001 "uuid-12345"

为什么 value 要用唯一值（UUID）？

场景：锁误删问题

T0:  线程 A 获得锁，value = "uuid-A"，TTL = 30s
T25: 线程 A 执行慢，锁快过期
T30: 锁过期，自动释放
T31: 线程 B 获得锁，value = "uuid-B"
T32: 线程 A 执行完，调用 DEL lock:order:1001
     → 如果不验证 value，会把线程 B 的锁删掉！❌
     → 验证 value == "uuid-A" 失败，不删除 ✅

正确流程：
  1. 加锁：SET lock:key "uuid-xxx" NX EX 30
  2. 解锁：Lua 脚本先 GET 验证 value，再 DEL
  3. 每个线程用不同的 UUID，防止误删其他线程的锁

完整代码示例：

import uuid
import redis

r = redis.Redis()

def acquire_lock(lock_key, ttl=30):
    """加锁"""
    identifier = str(uuid.uuid4())
    # SET key value NX EX seconds
    if r.set(lock_key, identifier, nx=True, ex=ttl):
        return identifier
    return None

def release_lock(lock_key, identifier):
    """解锁"""
    script = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else
        return 0
    end
    """
    r.eval(script, 1, lock_key, identifier)

# 使用
lock_key = "lock:order:1001"
identifier = acquire_lock(lock_key)
if identifier:
    try:
        # 执行业务逻辑
        process_order()
    finally:
        release_lock(lock_key, identifier)

性能数据：

测试场景：1000 次加锁 + 解锁
- 错误实现（SETNX + EXPIRE）：吞吐量约 2000 QPS，有死锁风险
- 正确实现（SET NX EX）：吞吐量约 5000 QPS，无死锁风险

Redis 单节点分布式锁性能：
  - 加锁 + 解锁：约 0.1ms（单次网络往返）
  - 吞吐量：约 10000 QPS（单 Redis 实例）

本质一句话：分布式锁用 SET key value NX EX 加锁，用 Lua 脚本验证 value 后解锁，value 用 UUID 防止误删其他线程的锁。

Q4：Redisson 的看门狗机制是什么？解决了什么问题？高频

回答要点：

问题：锁过期但业务未完成

场景：
T0:  线程 A 获得锁，TTL = 30s
T25: 线程 A 还在执行，但锁快过期
T30: 锁过期，线程 B 获得锁
T31: 线程 A 和线程 B 同时持有锁 ❌

问题：
  1. 锁的 TTL 固定，但业务执行时间不确定
  2. TTL 太短 → 业务未完成锁就过期，不安全
  3. TTL 太长 → 业务崩溃后锁长时间不释放，不可用

Redisson 的看门狗解决方案：

看门狗（Watchdog）机制：
  1. 获取锁时不指定 TTL，使用默认 30s（lockWatchdogTimeout）
  2. 获取锁成功后，启动后台线程（看门狗）
  3. 后台线程每隔 10s（timeout/3）检查锁是否还持有
  4. 如果持有，自动续期到 30s
  5. 业务完成，调用 unlock() 释放锁，看门狗自动停止

流程：
T0:  线程 A 获得锁，TTL = 30s，启动看门狗
T10: 看门狗检查：锁还持有 → 续期到 30s
T20: 看门狗检查：锁还持有 → 续期到 30s
T25: 线程 A 执行完，unlock() 释放锁
T26: 看门狗检测到锁已释放，自动停止

代码示例：

// Redisson 示例
RedissonClient redisson = Redisson.create(config);
RLock lock = redisson.getLock("lock:order:1001");

// 方式1：看门狗自动续期（不指定 leaseTime）
lock.lock();  // 默认 TTL = 30s，看门狗每 10s 续期
try {
    // 执行业务逻辑（可能超过 30s）
    processOrder();
} finally {
    lock.unlock();
}

// 方式2：手动指定 leaseTime（不启动看门狗）
lock.lock(60, TimeUnit.SECONDS);  // TTL = 60s，不续期
try {
    processOrder();
} finally {
    lock.unlock();
}

配置参数：

Config config = new Config();
config.setLockWatchdogTimeout(30000);  // 看门狗超时时间，默认 30s
// 续期间隔 = lockWatchdogTimeout / 3 = 10s

看门狗失效的场景：

场景1：JVM 崩溃（Full GC、OOM）
  看门狗线程停止 → 锁在 30s 后自动释放 ✅

场景2：服务器宕机
  看门狗随进程消失 → 锁在 30s 后自动释放 ✅

场景3：业务线程阻塞
  业务线程阻塞，但看门狗仍在运行 → 锁持续续期
  → 需要业务层监控，避免死锁 ⚠️

本质一句话：看门狗通过后台线程定时续期，解决”锁过期但业务未完成”的问题，业务崩溃后锁自动释放。

Q5：RedLock 算法是什么？有什么争议？中频

回答要点：

单节点分布式锁的问题：

场景：Redis 主从架构
T0:  线程 A 在主库获得锁
T1:  主库宕机，锁未同步到从库
T2:  从库提升为主库
T3:  线程 B 在新主库获得锁
→ 线程 A 和线程 B 同时持有锁 ❌

RedLock 算法（Antirez 提出）：

假设有 5 个独立 Redis 实例（非主从，完全独立）：

1. 记录开始时间 t1
2. 依次向 5 个实例发送 SET key value NX PX ttl
3. 记录结束时间 t2，计算获取锁耗时 elapsed = t2 - t1
4. 如果在超过半数（≥3）的实例上获取成功，且 elapsed < ttl，则加锁成功
5. 锁的有效时间 = ttl - elapsed（减去网络耗时）
6. 释放时，向所有实例发送释放脚本

示例：
  实例 1: 获取成功
  实例 2: 获取成功
  实例 3: 获取失败
  实例 4: 获取成功
  实例 5: 获取成功
  → 4/5 成功，加锁成功 ✅

RedLock 的争议（Martin Kleppmann 的批评）：

问题1：时钟漂移（Clock Skew）
  假设 ttl = 10s，实例 A 的时钟比实例 B 快 11s
  T0:  客户端在实例 A、B、C 获得锁
  T5:  实例 A 时钟快，认为锁已过期（T0+10s），删除锁
  T6:  客户端2 在实例 A 获得锁
  → 两个客户端同时持有锁 ❌

问题2：GC Pause（Stop-The-World）
  T0:  客户端1 获得 RedLock，ttl = 10s
  T5:  客户端1 JVM 触发 Full GC，暂停 15 秒
  T10: 锁过期，客户端2 获得锁
  T20: 客户端1 GC 结束，以为自己还持有锁，继续操作
  → 两个客户端同时操作，破坏互斥性 ❌

问题3：网络延迟
  获取锁的响应在网络中延迟，等客户端收到响应时锁已快过期
  → 实际有效期远小于 ttl

Antirez 的回应：

1. 时钟漂移：生产环境用 NTP 同步时钟，漂移极小
2. GC Pause：Java 应用可通过 -XX:+UnlockExperimentalVMOptions
   -XX:UseMaximumCompactionOnFullGC 减少 GC 暂停
3. 网络延迟：计算 elapsed 并减去网络耗时，已考虑此问题

结论：
  RedLock 提供了"足够好"的工程安全性
  极端场景概率极低，实践中可接受

工程实践：

对数据安全性要求极高的场景（如金融交易）：
  不要依赖分布式锁做最终保证
  → 用数据库唯一约束或乐观锁（版本号）兜底

对数据安全性要求一般的场景（如库存扣减）：
  RedLock 提供了足够的安全性
  → 结合幂等设计，避免重复扣减

本质一句话：RedLock 通过多节点投票提高安全性，但在时钟漂移、GC Pause 等极端场景下可能失效，需结合数据库约束兜底。

链式追问三：缓存穿透

Q6：什么是缓存穿透？如何解决？必考

回答要点：

缓存穿透现象：

场景：查询不存在的 key
恶意请求 id=-1 → Redis 未命中 → MySQL 未命中
→ 每次都穿透缓存，直达数据库
→ DB 被拖垮 ❌

正常请求：
  请求 → Redis 命中 → 返回（不查 DB）

穿透请求：
  请求 → Redis 未命中 → MySQL 未命中 → 返回 null
  → 下次请求仍穿透（因为 Redis 没缓存）

解决方案对比：

方案	实现方式	优点	缺点	适用场景
缓存空值	MySQL 未命中时，缓存 key=null，TTL 短	实现简单	Redis 内存占用高（大量空值）	查询 key 相对固定
布隆过滤器	启动时加载所有合法 ID 到布隆过滤器	内存占用低	有误判（假阳性）	合法 ID 集合相对固定

方案1：缓存空值：

def get_user(user_id):
    # 查缓存
    value = redis.get(f"user:{user_id}")
    if value is not None:
        if value == "NULL":  # 缓存的空值
            return None
        return value

    # 查数据库
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    if user:
        redis.set(f"user:{user_id}", user, ex=3600)
    else:
        # 缓存空值，TTL 较短（60s）
        redis.set(f"user:{user_id}", "NULL", ex=60)

    return user

方案2：布隆过滤器：

import pybloom_live

# 启动时：加载所有合法 ID 到布隆过滤器
bloom = pybloom_live.ScalableBloomFilter(
    initial_capacity=1000000,
    error_rate=0.001
)
for user_id in db.query("SELECT id FROM users"):
    bloom.add(user_id)

def get_user(user_id):
    # 先查布隆过滤器
    if user_id not in bloom:
        return None  # 一定不存在

    # 查缓存
    value = redis.get(f"user:{user_id}")
    if value:
        return value

    # 查数据库
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    if user:
        redis.set(f"user:{user_id}", user, ex=3600)

    return user

布隆过滤器原理：

布隆过滤器（Bloom Filter）：
  结构：一个大的位数组（如 1 亿 bit = 12.5MB）
  插入：用 k 个哈希函数计算 k 个位置，全部置 1
  查询：检查 k 个位置是否全为 1
    - 全为 1：可能存在（有假阳性）
    - 有一个为 0：一定不存在（无假阴性）

示例：
  插入元素 "user:1001"：
    hash1("user:1001") % n = 5 → bit[5] = 1
    hash2("user:1001") % n = 23 → bit[23] = 1
    hash3("user:1001") % n = 67 → bit[67] = 1

  查询元素 "user:9999"：
    hash1("user:9999") % n = 5 → bit[5] = 1 ✅
    hash2("user:9999") % n = 23 → bit[23] = 1 ✅
    hash3("user:9999") % n = 42 → bit[42] = 0 ❌
    → 元素一定不存在

性能对比：

测试场景：100 万个合法 ID，1000 万次查询（其中 80% 是非法 ID）
- 无防护：DB QPS = 1000 万次 × 80% = 800 万次 → DB 崩溃
- 缓存空值：Redis 内存占用约 100MB（100 万个空值）
- 布隆过滤器：Redis 内存占用约 2MB（位数组），误判率 0.1%

结论：布隆过滤器内存占用低，但有少量误判（0.1% 的非法请求会穿透到 DB）。

本质一句话：缓存穿透用”缓存空值”或”布隆过滤器”解决，前者简单但占内存，后者内存省但有少量误判。

链式追问四：缓存击穿

Q7：什么是缓存击穿？如何解决？必考

回答要点：

缓存击穿现象：

场景：热点 key 突然过期
正常：请求 → Redis 命中热点 key → 返回
过期瞬间：10000 QPS 全部 → Redis 未命中 → 10000 个请求同时查 DB → DB 崩溃 ❌

区别：
  - 缓存穿透：查询不存在的 key
  - 缓存击穿：热点 key 过期
  - 缓存雪崩：大量 key 同时过期

解决方案对比：

方案	实现方式	优点	缺点
互斥锁（Mutex）	只允许一个线程重建缓存	数据强一致	其他线程等待，有延迟
逻辑过期	不设 TTL，数据中存过期时间	请求不阻塞	短时间返回旧数据

方案1：互斥锁：

def get_with_mutex(key):
    value = redis.get(key)
    if value:
        return value

    # 尝试获取互斥锁
    lock_key = f"lock:{key}"
    if redis.set(lock_key, 1, nx=True, ex=10):  # 锁 TTL = 10s
        try:
            # 查数据库
            value = db.query(key)
            # 写入缓存
            redis.set(key, value, ex=300)
            return value
        finally:
            redis.delete(lock_key)
    else:
        # 未获得锁，等待 50ms 后重试
        time.sleep(0.05)
        return get_with_mutex(key)

方案2：逻辑过期：

import json
import time
import threading

def get_with_logical_expire(key):
    obj_str = redis.get(key)
    if not obj_str:
        # 缓存未命中，加载数据
        return load_and_cache(key)

    # 解析数据：{"expire_time": 1704067200, "data": {...}}
    obj = json.loads(obj_str)
    if obj["expire_time"] > time.time():
        return obj["data"]  # 未逻辑过期

    # 已逻辑过期，尝试获取锁异步重建
    lock_key = f"lock:{key}"
    if redis.set(lock_key, 1, nx=True, ex=10):
        # 启动异步线程重建缓存
        threading.Thread(target=rebuild_cache, args=(key,)).start()

    # 无论是否获得锁，返回旧数据（保证可用性）
    return obj["data"]

def rebuild_cache(key):
    """异步重建缓存"""
    value = db.query(key)
    obj = {
        "expire_time": time.time() + 300,  # 逻辑过期时间 = 当前时间 + 5 分钟
        "data": value
    }
    redis.set(key, json.dumps(obj))  # 不设 Redis TTL
    redis.delete(f"lock:{key}")

def load_and_cache(key):
    """首次加载数据"""
    value = db.query(key)
    obj = {
        "expire_time": time.time() + 300,
        "data": value
    }
    redis.set(key, json.dumps(obj))
    return value

两种方案对比：

互斥锁方案：
  T0: 线程 A 发现缓存过期，获得锁，查 DB
  T1: 线程 B、C、D... 发现缓存过期，未获得锁，等待
  T2: 线程 A 重建缓存完成，释放锁
  T3: 线程 B、C、D... 重新查询，命中缓存
  → 数据强一致，但有等待延迟

逻辑过期方案：
  T0: 线程 A 发现逻辑过期，获得锁，启动异步重建
  T1: 线程 A、B、C... 都返回旧数据（逻辑过期但 Redis 未过期）
  T2: 异步线程重建缓存完成
  T3: 后续请求命中新缓存
  → 请求不阻塞，但短时间返回旧数据

性能对比：

测试场景：热点 key 过期瞬间，10000 QPS 打到 DB
- 无防护：DB QPS = 10000，DB 崩溃 ❌
- 互斥锁：DB QPS = 1（只有一个线程查 DB），其他线程等待约 50ms
- 逻辑过期：DB QPS = 1（异步查 DB），请求无等待

结论：
  - 数据强一致场景 → 互斥锁
  - 高可用场景（允许短时间脏读）→ 逻辑过期

本质一句话：缓存击穿用互斥锁（强一致）或逻辑过期（高可用）解决，前者让请求等待，后者返回旧数据。

链式追问五：缓存雪崩

Q8：什么是缓存雪崩？如何解决？必考

回答要点：

缓存雪崩现象：

场景1：大量 key 同时过期
  T0:  10000 个 key 同时过期（如都设置了 1 小时 TTL）
  T1:  10000 个请求同时穿透到 DB
  → DB QPS 飙升，崩溃 ❌

场景2：Redis 服务宕机
  T0:  Redis 宕机
  T1:  所有请求穿透到 DB
  → DB QPS 飙升，崩溃 ❌

解决方案：

方案1：TTL 随机抖动
  避免大量 key 同时过期：
    ❌ redis.set(key, value, ex=3600)  # 整点过期
    ✅ redis.set(key, value, ex=3600 + random.randint(0, 600))  # 随机抖动

方案2：多级缓存（Local Cache + Redis）
  请求 → 本地缓存（Caffeine，毫秒级）
           ↓ 未命中
         Redis 缓存（毫秒级）
           ↓ 未命中
         数据库

  优点：
    - Redis 宕机时，本地缓存仍可抗住大部分流量
    - 减少对 Redis 的压力

方案3：熔断降级
  Redis 不可用时：
    - 熔断器打开，直接返回降级响应（兜底数据）
    - 防止 DB 被雪崩压垮

方案4：Redis 高可用
  - 主从 + 哨兵：自动故障转移
  - Redis Cluster：分片 + 高可用

代码示例：

# 方案1：TTL 随机抖动
import random

def set_cache(key, value, ttl=3600):
    jitter = random.randint(0, int(ttl * 0.1))  # 随机抖动 0~10%
    redis.set(key, value, ex=ttl + jitter)

# 方案2：多级缓存
from cachetools import TTLCache

local_cache = TTLCache(maxsize=10000, ttl=60)  # 本地缓存，TTL = 60s

def get_user(user_id):
    # 先查本地缓存
    if user_id in local_cache:
        return local_cache[user_id]

    # 查 Redis 缓存
    value = redis.get(f"user:{user_id}")
    if value:
        local_cache[user_id] = value  # 写入本地缓存
        return value

    # 查数据库
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    if user:
        set_cache(f"user:{user_id}", user)
        local_cache[user_id] = user

    return user

# 方案3：熔断降级（使用 Hystrix 或 Resilience4j）
from resilience4j.circuitbreaker import CircuitBreaker

circuit_breaker = CircuitBreaker.of_defaults("redis")

@circuit_breaker.decorate
def get_user_with_fallback(user_id):
    return get_user(user_id)

def get_user(user_id):
    try:
        return get_user_with_fallback(user_id)
    except Exception:
        # 熔断后返回降级数据
        return get_default_user()

性能对比：

测试场景：Redis 宕机，10000 QPS 打到 DB
- 无防护：DB QPS = 10000，DB 崩溃 ❌
- 多级缓存：本地缓存命中率 80%，DB QPS = 2000，DB 可承受 ✅
- 熔断降级：熔断器打开，DB QPS = 0，直接返回降级数据 ✅

测试场景：10000 个 key 同时过期
- TTL 无抖动：DB QPS 飙升到 10000，DB 崩溃 ❌
- TTL 随机抖动：DB QPS 分散到 10 分钟内，峰值约 200，DB 正常 ✅

本质一句话：缓存雪崩用 TTL 随机抖动、多级缓存、熔断降级解决，核心是避免大量请求同时打到 DB。

链式追问六：缓存与数据库一致性

Q9：如何保证缓存和数据库的一致性？必考

回答要点：

根本矛盾：

缓存和数据库是两个独立系统，无法做到原子更新：

场景：更新数据
  方案1：先更缓存，再更 DB
    T1: 更新缓存 key=100
    T2: 数据库更新失败
    → 缓存和数据库不一致 ❌

  方案2：先更 DB，再更缓存
    T1: 更新数据库 key=100
    T2: 更新缓存失败
    → 缓存和数据库不一致 ❌

常见方案对比：

方案	流程	优点	缺点
先更 DB，再删缓存	UPDATE DB → DELETE cache	缓存下次读时重建，避免并发写	不一致窗口极小
先更 DB，再更缓存	UPDATE DB → SET cache	下次读命中缓存	并发写时可能缓存旧值
先删缓存，再更 DB	DELETE cache → UPDATE DB	简单	读请求可能读到旧值并回写缓存
Canal + binlog	订阅 binlog 异步删缓存	与业务代码解耦	需要额外组件

方案1：先更新 DB，再删除缓存（推荐）：

def update_user(user_id, data):
    # 1. 更新数据库
    db.execute(f"UPDATE users SET name = '{data['name']}' WHERE id = {user_id}")

    # 2. 删除缓存（不更新缓存）
    redis.delete(f"user:{user_id}")

    # 为什么删缓存而不是更新缓存？
    # - 并发写时，删除缓存可避免缓存旧值
    # - 下次读时重建缓存，保证数据最新

为什么删缓存而不是更新缓存？

场景：并发写
T1: 线程 A 更新 DB（val=100）
T2: 线程 B 更新 DB（val=200）
T3: 线程 B 更新缓存（val=200）
T4: 线程 A 更新缓存（val=100）← 后写的反而是旧值 ❌
→ 缓存中存了旧值

如果删除缓存：
T1: 线程 A 更新 DB（val=100）
T2: 线程 B 更新 DB（val=200）
T3: 线程 B 删除缓存
T4: 线程 A 删除缓存
→ 下次读时重建缓存，数据正确 ✅

潜在的不一致窗口：

场景：读请求在写请求删除缓存前读到旧数据并回写
T1: 线程 A 更新 DB（val=100）
T2: 线程 B 读缓存未命中，读 DB（读到旧值 val=50）
T3: 线程 A 删除缓存
T4: 线程 B 将旧值（val=50）写入缓存
→ 缓存中存了旧数据，直到 TTL 过期

概率：
  数据库写操作（约 50ms）+ 读操作（约 5ms）= 55ms
  读请求在这 55ms 内命中不一致窗口的概率极低

方案2：延迟双删：

def update_user(user_id, data):
    # 第一次删除缓存
    redis.delete(f"user:{user_id}")

    # 更新数据库
    db.execute(f"UPDATE users SET name = '{data['name']}' WHERE id = {user_id}")

    # 延迟 100ms 后第二次删除（兜底）
    time.sleep(0.1)
    redis.delete(f"user:{user_id}")

# 解决不一致窗口问题，但 sleep 影响性能

方案3：Canal + binlog（强一致方案）：

架构：
  MySQL → binlog → Canal → 解析变更 → 删除/更新 Redis

优点：
  - 异步，与业务代码解耦
  - 延迟通常在毫秒级（1~10ms）
  - 保证最终一致性

缺点：
  - 需要额外维护 Canal 组件
  - 引入新的复杂性

本质一句话：缓存一致性用”先更 DB，再删缓存”实现，不一致窗口极小；强一致用 Canal + binlog，但有组件复杂度。

总结

核心要点回顾

Redis 原子性：单命令和 Lua 脚本原子，MULTI/EXEC 不支持回滚
分布式锁：SET NX EX 加锁，Lua 脚本解锁，value 用 UUID 防误删，Redisson 看门狗续期
RedLock 争议：多节点投票提高安全性，但时钟漂移、GC Pause 等极端场景可能失效
缓存穿透：缓存空值或布隆过滤器，防止大量请求打到 DB
缓存击穿：互斥锁（强一致）或逻辑过期（高可用）
缓存雪崩：TTL 随机抖动、多级缓存、熔断降级
缓存一致性：先更 DB 再删缓存，不一致窗口极小；强一致用 Canal + binlog

面试高频考点

<Badge text="必考" variant="danger" /> Redis 如何保证原子性？MULTI/EXEC 和数据库事务区别？
<Badge text="必考" variant="danger" /> 分布式锁如何实现？为什么 value 要用 UUID？
<Badge text="必考" variant="danger" /> 缓存穿透、击穿、雪崩的区别和解决方案？
<Badge text="必考" variant="danger" /> 如何保证缓存和数据库的一致性？
<Badge text="高频" variant="tip" /> Redisson 的看门狗机制？
<Badge text="高频" variant="tip" /> Lua 脚本的优势和注意事项？
<Badge text="中频" variant="note" /> RedLock 算法原理和争议？
<Badge text="实战" variant="caution" /> 实际项目中如何防止缓存穿透/击穿/雪崩？