第 11 章 分布式锁（含 Redlock 争议）

学习目标：彻底搞清楚「为什么单机锁在多机环境会失效」「Redis 分布式锁正确写法」「为什么释放锁要用 Lua」「看门狗续期怎么做」「Redlock 到底安不安全」这五件大事。看完之后能解释 Martin Kleppmann 与 antirez 的著名辩论，并在面试里给出选型建议。

---

11.1 为什么需要分布式锁

🚽 生活类比：合租房有 1 个卫生间、3 个室友，谁要进去就把钥匙拿走，门口挂上「使用中」。等他出来，把钥匙挂回去，下一个人才能拿。

• 单机程序：钥匙就放在桌上（进程内的一把 mutex），三个室友共用一张桌子，看一眼就知道钥匙在不在。
• 多机程序：三个室友住三套不同的房子，桌子上的钥匙各自看不到——这时就需要一个所有人都能看到的钥匙挂钩（共享存储），这个钥匙挂钩就是「分布式锁」。

11.1.1 单机 synchronized 在多机环境失效

单机部署（Node-1 内 4 个线程抢资源）：
┌────────────────────────────────────┐
│  JVM 进程                          │
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │
│  │T1    │ │T2    │ │T3    │ │T4    │ │
│  └──┬───┘ └──┬───┘ └──┬───┘ └──┬───┘ │
│     └─────synchronized(lock)──────┘ │   ← JVM 内的对象锁，全员可见
└────────────────────────────────────┘
            ✅ 同一时刻只有一个线程进入临界区

多机部署（Node-1 / Node-2 / Node-3 各跑一个 JVM）：
┌────────┐  ┌────────┐  ┌────────┐
│ Node-1 │  │ Node-2 │  │ Node-3 │
│ lockA  │  │ lockB  │  │ lockC  │   ← 三把不同的 JVM 内对象，互相看不见
└───┬────┘  └───┬────┘  └───┬────┘
    └───────扣库存─────────┘
            ❌ 三个进程各扣各的，超卖！

核心矛盾：synchronized / ReentrantLock 这些工具锁住的是进程内的内存对象，多个进程的对象天然不共享。

11.1.2 解决思路：找一个「公共瞭望塔」

谁都能看见，谁都能写入，谁都能读 —— 把锁的状态放在这种地方：

┌────────┐  ┌────────┐  ┌────────┐
│ Node-1 │  │ Node-2 │  │ Node-3 │
└───┬────┘  └───┬────┘  └───┬────┘
    └───┐       │       ┌───┘
        ▼       ▼       ▼
       ┌──────────────────┐
       │   公共瞭望塔     │  ← Redis / Zookeeper / etcd / DB ...
       │  lock = "node1"  │
       └──────────────────┘

11.1.3 三种主流实现对比

维度	Redis（SET NX）	Zookeeper（临时顺序节点）	etcd（Lease + Revision）
CAP 倾向	AP（高可用，弱一致）	CP（强一致，牺牲可用性）	CP（基于 Raft）
性能	⭐⭐⭐⭐⭐（10W+ QPS）	⭐⭐（写需多数派同步）	⭐⭐⭐
加锁延迟	1~3 ms	10~50 ms	5~20 ms
客户端宕机检测	靠 TTL 过期（粗）	心跳断开立刻释放（精）	Lease 心跳
可重入	自己实现	自己实现（节点内计数）	自己实现
公平性	默认无；Redisson 提供 FIFO	临时顺序节点天然 FIFO	Watch 机制可实现
实现复杂度	简单（一行命令）	中（要会 Curator）	中
典型场景	高并发、能容忍极端边界	强一致（金融、配置中心）	K8s 控制器选主

🎯 选型口诀：

• 一般业务（90% 场景）：Redis——简单、快、生态成熟。
• 强一致、不容许双主：Zookeeper / etcd——慢一点但稳。
• 极端关键路径（钱）：分布式锁只做「优化」，业务侧仍要做幂等 + 唯一约束兜底。

---

11.2 单机 Redis 锁的演进史

⚠️ 接下来三个版本是「踩坑录」。只看 V3 的同学也请回头读 V1/V2 ——面试官最喜欢追问「为什么不用 SETNX + EXPIRE」。

11.2.1 V1：SETNX + DEL（致命缺陷：客户端崩了锁永远不释放）

# 加锁
SETNX my_lock 1     # 不存在则设置为 1，返回 1（成功）；存在返回 0（失败）

# 业务……

# 释放锁
DEL my_lock

时序问题：

时间线 ──►
Client-A:   SETNX (成功) ──► 业务执行 ──► 💥 进程崩溃 / kill -9
                                                 ↑
                                          锁还在，永远不会被 DEL！

Client-B:   等待 ──► SETNX (一直失败) ──► 等待 ──► 等待 ──► 死锁
Client-C:   等待 ──► SETNX (一直失败) ──► ...

结论：V1 没有过期保护，一旦持锁者崩溃，全集群死锁。

11.2.2 V2：SETNX + EXPIRE 分两步（致命缺陷：两步之间崩了仍死锁）

SETNX my_lock 1
EXPIRE my_lock 30       # 设置过期时间 30 秒

时序问题：

时间线 ──►
Client-A:   SETNX (成功) ──► 💥 网络抖动 / 进程崩 ──► （没来得及 EXPIRE）
                                                       ↑
                                              锁存在但没有 TTL，永久死锁

Client-B:   等待…等待…等待…等待…等待…等待…

问题本质：两条命令之间不是原子的。即使你写在 try-finally 里，应用进程被 kill -9 也救不了。

💡 历史上有人用 MULTI/EXEC 把这两条命令包起来「原子化」——但 SETNX 不能基于 MULTI 内的结果决定要不要 EXPIRE，所以这条路也走不通。

11.2.3 V3：SET key value NX EX 30 —— 一条命令搞定（推荐 ✅）

Redis 2.6.12 起，SET 命令扩展了选项，可以原子完成「不存在则设置 + 同时设过期」：

SET my_lock <unique_token> NX EX 30
#           ─────────────  ──  ────
#               值         不存在则设置  过期时间 30 秒

NX 与 EX 在同一条命令中由 Redis 单线程原子执行，不可能在中间被打断。

时间线 ──►
Client-A:   SET ... NX EX 30 (成功) ──► 业务 ──► 💥 崩溃
                                                      │
                            30 秒后 ──► Redis 自动删除 ──► 锁释放
Client-B:   等待 ──► 30 秒后 SET ... NX EX 30 (成功) ──► 业务

✅ 这就是「单机 Redis 分布式锁」的最小正确实现。后面要讨论的所有问题（释放校验、续期、可重入、公平性、Redlock）都是在 V3 基础上打补丁。

11.2.4 时序问题可视化

                    V1                       V2                      V3
                  ────                     ────                    ────
T0   ClientA:  SETNX OK              SETNX OK                  SET NX EX OK
T1              │                     │                          │
T2              │ 业务...              │ 💥 (在 EXPIRE 前崩了)      │ 业务...
T3              │                                                 │
T4              💥                    （锁无 TTL，永久占用）       │
T5                                                                ⏰ 30s 到期，自动释放
T∞    死锁，需要人工干预              死锁，需要人工干预           ✅ 自愈

---

11.3 锁的 5 个核心问题

V3 只是「能用」，离「生产可用」还差很远。下面 5 个问题是面试必考。

11.3.1 问题 1：锁要有过期时间（避免死锁）

V3 的 EX 30 已经解决。关键提醒：

• 过期时间设多大？经验值：业务平均执行时间的 2~3 倍。
• 太短：业务没跑完锁就被释放，别人能进来 → 数据不一致。
• 太长：故障恢复慢，崩了之后所有人等几分钟。

11.3.2 问题 2：释放锁要校验持有者（避免误删别人的锁）

反例：直接 DEL my_lock 有什么问题？

T0  ClientA: SET lock A_token NX EX 10  → OK，A 拿到锁
T1  ClientA: 业务执行……
T11 (10s 后)：锁过期自动释放
T12 ClientB: SET lock B_token NX EX 10  → OK，B 拿到锁
T13 ClientA: 业务终于跑完 ──► DEL lock     ← 💥 把 B 的锁删了！
T14 ClientC: SET lock C_token NX EX 10  → OK，C 也进来了
                                              （现在 B 和 C 都以为自己持锁）

根因：A 释放的是「锁」这个 key，但它不知道当前持锁的是不是自己。

解法：加锁时写入唯一 token（UUID），释放前校验。校验 + 删除必须原子，否则又会出现「校验通过后、删除前」被别人抢走的情况。

-- 释放锁的 Lua 脚本（保证原子）
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
else
    return 0
end

调用：

EVAL <上面的脚本> 1 my_lock A_token

💡 为什么必须 Lua？ Redis 单线程模型保证了一段 Lua 脚本中所有命令原子执行，相当于把 GET + DEL 拼成「一条」命令，中间不会被任何其他客户端打断。

11.3.3 问题 3：锁的「续命」问题（业务执行超过过期时间怎么办）

设过期 30s
T0:  A 获取锁
T0~T29: A 在执行业务（遇到了 GC / 慢 SQL / 第三方接口慢）
T30: 锁过期，B 进来拿到锁
T35: A 业务终于跑完 ──► 试图释放 → Lua 校验不通过 → 不会误删 ✅
                                  但 A 和 B 同时进了临界区 ❌

问题本质：Redis 不知道你的业务有没有跑完，只能机械地按 TTL 删除。

解法：看门狗（Watchdog）守护线程

        加锁线程                          看门狗线程（守护）
        ────────                          ──────────────
T0      SET lock NX PX 30000  ──── 启动 ──→ 每 10s 检查一次
                                              ↓
T10                                    GET lock == my_token?
                                              ↓ 是
                                         PEXPIRE lock 30000  ← 续到 30s
T20                                    GET == my_token? → PEXPIRE 30s
T30                                    GET == my_token? → PEXPIRE 30s
T35     业务跑完 → 释放锁 → 通知看门狗停止

                持锁                续期                                    释放
锁的 TTL: ─►|━━━━30s━━━━|━━━━30s━━━━|━━━━30s━━━━|━━━━30s━━━━|━━━━━━━━×
                ↑           ↑           ↑           ↑           ↑
                T0(获取)   T10(续)    T20(续)    T30(续)    T35(释放并停看门狗)

⚠️ 三个细节：

1. 续期前必须校验是不是自己的锁（同样用 Lua）。否则锁过期被别人拿走后，A 还在傻傻续期 → 抢走别人的锁。
2. 续期间隔 = 过期时间 / 3 ~ /2 是经验值。
3. 看门狗必须在主线程释放锁后及时停止，避免泄漏。

Java 生态的 Redisson 已经把看门狗做成默认能力，详见 11.5。

11.3.4 问题 4：可重入（同一线程多次获取）

「重入」= 同一个线程已经持锁，再次申请同一把锁应当成功，且释放要平衡。

A.lock()       ─→ count=1  （获取，count++）
A.lock()       ─→ count=2  （重入，count++）
   doSomething()
A.unlock()     ─→ count=1  （还有一层，不真删）
A.unlock()     ─→ count=0  （平衡到 0，DEL key）

为什么需要可重入？ 同一个线程递归调用，或者 A 函数加锁后调用了 B 函数，B 函数也想加同一把锁——如果不可重入，就会自己把自己锁死。

Redisson 用 Hash 实现：

HASH key=my_lock
  ┌────────────────────────────────────────────┐
  │ field = "uuid:thread-id"   value = "count" │
  └────────────────────────────────────────────┘

加锁脚本（伪代码）：
  if not exists(key): HSET key field 1; PEXPIRE key 30000; return OK
  if HEXISTS(key, field): HINCRBY key field 1; PEXPIRE key 30000; return OK
  else: return PTTL key   -- 已被别人持有，返回剩余时间

解锁脚本：
  if not HEXISTS(key, field): return null
  count = HINCRBY key field -1
  if count > 0: PEXPIRE key 30000; return 0
  else:        DEL key; return 1

11.3.5 问题 5：公平性（FIFO 排队）

默认的 SET NX 是「抢占式」：100 个客户端同时 SET NX，谁手快谁拿到，先来的不一定先拿。

某些场景需要严格的 FIFO（避免饥饿）：

不公平：             公平（FIFO）：
A 等 0.1s 拿到       A 等 0.1s 拿到（第 1 名进队）
B 等 5s 拿到         B 等 0.2s 进队 → 等 A 释放 → 拿到（第 2 名）
C 等 20s（饥饿！）   C 等 0.3s 进队 → 等 B 释放 → 拿到

实现思路：

方案	数据结构	思路
Redisson FairLock	List（thread queue） + ZSet（超时）	进队后 SUBSCRIBE 等通知，按队列顺序唤醒
基于 Stream	XADD 进队 + XREADGROUP	类似消息队列
基于 ZSet	ZADD 时间戳 → ZRANGE 0 0 取队首	简单，需轮询

⚠️ 公平锁吞吐量比非公平锁低 30%~50%。只有在饥饿不可接受时才用。

---

11.4 Redlock 算法（多实例分布式锁）

11.4.1 为什么单实例不够？

单实例 Redis（即使加 slave 主从）有一个致命窗口：

T0   ClientA: SET lock A NX EX 30  → master OK
T1   master 把命令通过异步复制发往 slave  ──► 在路上
T2   master 突然挂掉，sentinel 把 slave 升为新 master
                                     ↑
                       新 master 上没有 lock 这个 key（复制还没到）
T3   ClientB: SET lock B NX EX 30  → 新 master OK
                                     ↑
                              💥 A 和 B 同时持有锁！

根因：Redis 主从复制是异步的，故障切换时未同步的写会丢失。

11.4.2 Redlock 的核心想法

antirez 在 2016 年提出 Redlock：与其相信「一个 Redis 集群」，不如部署 N 个完全独立的 Redis 实例（不是主从！），向多数派申请锁，达到 N/2+1 才算成功。

                 ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
                 │ Redis-1│ │ Redis-2│ │ Redis-3│ │ Redis-4│ │ Redis-5│
                 └────┬───┘ └────┬───┘ └────┬───┘ └────┬───┘ └────┬───┘
                      │          │          │          │          │
   Client ───依次 SET NX EX────►   ►          ►          ►          ►
                      ✓          ✓          ✓          ✗          ✗
                          ↑ 5 个里成功 3 个 (≥3 = N/2+1) → 加锁成功

⚠️ 关键约束：N 个实例必须是完全独立的（不同机器、不同电源、不同机架），不能是主从——主从同时挂一份就完了。常见取 N=5。

11.4.3 算法步骤

┌──────────────────────────────────────────────────────────────────┐
│                     Redlock 加锁完整流程                            │
├──────────────────────────────────────────────────────────────────┤
│ Step 1: 客户端记录当前毫秒时间 t1                                   │
│                                                                    │
│ Step 2: 依次向 N 个 Redis 实例请求加锁                              │
│         SET lock my_uuid NX PX 30000                               │
│         （每个实例的请求超时设置很小，如 50ms                        │
│          —— 单个实例阻塞不能拖累整体）                              │
│                                                                    │
│ Step 3: 客户端记录当前毫秒时间 t2，计算耗时 Δt = t2 - t1            │
│                                                                    │
│ Step 4: 同时满足 → 加锁成功                                         │
│         (a) 在 N/2+1 个实例上加锁成功                              │
│         (b) Δt < 锁的过期时间                                      │
│              （否则锁的「真实有效时间」太短，没意义）                  │
│                                                                    │
│ Step 5: 锁的实际有效时间 = 设定 TTL - Δt                           │
│         （扣掉申请阶段已经消耗的时间）                                │
│                                                                    │
│ Step 6: 失败时（不足多数派 / 总耗时超过 TTL），                      │
│         向所有 N 个实例（包括没成功的）发释放命令                     │
│         —— 万一是网络抖动导致 client 没收到 ack，实际上已加锁          │
└──────────────────────────────────────────────────────────────────┘

时序图：N=5，TTL=30s
─────────────────────────────────────────────────
t1=0ms    Client → R1: SET ... → OK (5ms)
t1=5ms    Client → R2: SET ... → OK (4ms)
t1=9ms    Client → R3: SET ... → 超时 (50ms)   ← 不堵死
t1=59ms   Client → R4: SET ... → OK (3ms)
t1=62ms   Client → R5: SET ... → OK (2ms)
t2=64ms   Δt = 64ms
          多数派 4/5 ✅；Δt(64ms) < TTL(30000ms) ✅
          实际有效时间 = 30000 - 64 = 29936ms
─────────────────────────────────────────────────

11.4.4 著名争议：Kleppmann vs antirez

2016 年 2 月，Martin Kleppmann（《Designing Data-Intensive Applications》作者）发文 How to do distributed locking 怒喷 Redlock，antirez（Redis 作者）当晚回应 Is Redlock safe?，引发分布式系统圈一场大辩论。

Kleppmann 三大攻击：

┌────────────────────────────────────────────────────────────────┐
│ 攻击 1: GC 暂停                                                  │
├────────────────────────────────────────────────────────────────┤
│ Client A 加锁 OK                                                │
│   ↓                                                            │
│ Client A 进入 STW GC（Java 一次 Full GC 可达几秒到几十秒）         │
│   ↓                                                            │
│ TTL 期间 Client A 完全冻结，锁过期                                 │
│   ↓                                                            │
│ Client B 加锁 OK，开始操作资源                                    │
│   ↓                                                            │
│ Client A GC 结束，"以为自己"还持锁，开始操作资源                    │
│   ↓                                                            │
│ 💥 双写冲突，Redlock 帮不了你                                     │
└────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────┐
│ 攻击 2: 时钟漂移                                                 │
├────────────────────────────────────────────────────────────────┤
│ Redlock 算「过期」依赖每个实例本地时钟。                            │
│ 如果 NTP 校时把时钟跳跃了几秒（或运维手动改时间），                  │
│ 锁可能在多个实例上「同时」提前过期，多数派被破坏。                    │
└────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────┐
│ 攻击 3: 网络分区 + 进程暂停                                        │
├────────────────────────────────────────────────────────────────┤
│ Client A 持锁后被网络隔离/进程冻结，                                │
│ 锁过期 → Client B 拿到锁 → 双客户端同时认为自己持锁。                │
└────────────────────────────────────────────────────────────────┘

Kleppmann 的核心论点：任何基于 TTL 的锁都不能保证互斥，因为「客户端 A 还以为自己持锁」是 Redlock 检测不到的。要绝对正确，需要 fencing token：

fencing token 思路：
   每次成功加锁返回一个单调递增的版本号（token）
   Client A 拿到 token=33 → 业务 → 写存储时带 token
   Client A 卡住后 token 还是 33
   Client B 拿到 token=34 → 写存储 → 存储记下「最新见过 34」
   Client A 复活后写存储带 33 → 存储发现 33 < 34，拒绝写入 ✅

antirez 的反驳：

1. Kleppmann 用「假设系统 X 必须满足 Y」来评判 Redlock，标准过严。
2. Redlock 已经是 effort 上限：在 AP 系统约束下做到了「TTL 内多数派互斥」，业务想要更强保证，应该结合 fencing token，而不是怪 Redlock。
3. GC 暂停问题对 ZK 同样存在（ZK 的 session 也会因为 GC 失效）。
4. 时钟漂移：现代 NTP + 监控可控，不要用 NTP slewing 之外的方式调时间。

11.4.5 应不应该用 Redlock？

┌─────────────────────────────────────────────────────────────────┐
│ 决策树：                                                          │
│                                                                  │
│  你的锁保护的资源「绝对不能双写」吗？                                │
│        │                                                         │
│        ├── 是（金融、唯一性约束、计费扣款）                          │
│        │        ↓                                                │
│        │   不要只靠 Redlock                                       │
│        │   → 用 ZooKeeper / etcd + fencing token                  │
│        │   → 业务层用唯一索引、CAS、幂等表兜底                      │
│        │                                                         │
│        └── 否（限流、互斥任务调度、避免重复发邮件）                   │
│                 ↓                                                │
│            单 Redis 已足够（V3 + 看门狗）                          │
│            Redlock 反而增加运维复杂度（5 个独立实例）                │
└─────────────────────────────────────────────────────────────────┘

💡 个人结论（仅供参考）：日常业务用 单 Redis 主从 + V3 + Redisson 看门狗就够了；真正不能错的资源永远依赖业务幂等而不是锁。Redlock 是「学院派」的有趣设计，但很少有团队真的部署 5 套独立 Redis 来跑它。

---

11.5 Redisson 客户端实战

Redisson 是 Java 生态中最完善的 Redis 客户端，默认实现了上面所有补丁，开箱即用。

11.5.1 看门狗机制

默认：
  - 锁过期：30 秒（lockWatchdogTimeout）
  - 续期间隔：每 10 秒（= timeout / 3）
  - 续期前校验持有者
  - 释放锁时自动停止看门狗

注意：
  - 仅当不指定 leaseTime 时启用
  - 一旦你 lock(10, TimeUnit.SECONDS)，就关闭看门狗（按你说的来）

11.5.2 Java 风格伪代码

// 单机 Redisson
Config config = new Config();
config.useSingleServer().setAddress("redis://127.0.0.1:6379");
RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("order:1001");

// 写法 1：用看门狗（推荐）
lock.lock();   // 阻塞直到拿到锁，30s 自动续期
try {
    // 业务（即使跑 5 分钟也不会被别人抢走）
    decrementStock();
} finally {
    lock.unlock();   // 通知看门狗停止
}

// 写法 2：固定超时（不续期）
if (lock.tryLock(3, 10, TimeUnit.SECONDS)) {
    try { /* 业务，必须 10s 内完成 */ }
    finally { lock.unlock(); }
}

// 写法 3：Redlock（多实例）
RedissonClient r1 = Redisson.create(c1);
RedissonClient r2 = Redisson.create(c2);
RedissonClient r3 = Redisson.create(c3);
RedissonClient r4 = Redisson.create(c4);
RedissonClient r5 = Redisson.create(c5);

RLock l1 = r1.getLock("order:1001");
RLock l2 = r2.getLock("order:1001");
RLock l3 = r3.getLock("order:1001");
RLock l4 = r4.getLock("order:1001");
RLock l5 = r5.getLock("order:1001");

RedissonRedLock redLock = new RedissonRedLock(l1, l2, l3, l4, l5);
redLock.lock();
try { /* 业务 */ } finally { redLock.unlock(); }

11.5.3 看门狗续期源码思路

// 简化版（实际见 RedissonLock#scheduleExpirationRenewal）
private void scheduleExpirationRenewal(long threadId) {
    Timeout task = commandExecutor.getServiceManager().newTimeout(timeout -> {
        // 用 Lua 校验持有者后再 PEXPIRE，原子
        renewExpirationAsync(threadId).onComplete((res, e) -> {
            if (res) {
                // 续期成功，再次调度（递归）
                scheduleExpirationRenewal(threadId);
            }
            // 续期失败说明锁已不在自己手上，放弃续期
        });
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    // ↑ 关键：每 1/3 lease 续一次
}

💡 Python 生态对应库：python-redis-lock（带看门狗）、redlock-py（实现 Redlock 算法）。

---

11.6 5 个易错点 ⚠️

11.6.1 用 SETNX 不带过期 → 死锁

# ❌ 错
SETNX lock 1
# ... 业务崩溃 ...
# 锁永远不被释放

# ✅ 对
SET lock <token> NX EX 30

11.6.2 用 DEL 不校验持有者 → 释放别人的锁

# ❌ 错
if lock_held:
    redis.delete("lock")     # 可能你的锁早过期了，别人正在持有

# ✅ 对（用 Lua）
script = """
if redis.call("GET", KEYS[1]) == ARGV[1] then
    return redis.call("DEL", KEYS[1])
end
return 0
"""
redis.eval(script, 1, "lock", my_token)

11.6.3 业务超过过期时间但没续期 → 误删后被别人拿到

症状：业务跑了 35s，锁 TTL 30s，A 跑完去释放时锁已经被 B 拿走。

解法：上看门狗（11.3.3）。或者：把 TTL 设到业务 95% 分位的 3 倍以上。

11.6.4 主从异步复制丢失 → 双客户端同时获取

症状：master 设锁后没等同步到 slave 就崩了，slave 升主后锁丢失，B 也能加锁。

解法：

• 一般业务：接受这个极小概率，靠业务幂等兜底。
• 严肃场景：用 Redlock（多独立实例）或换 ZK。

11.6.5 时钟回拨 → Redlock 失效

症状：运维手动 date -s 把时间改回去几秒，多个 Redis 实例同时「认为锁过期了」。

解法：

• 禁止 step adjust，只允许 NTP slew（缓慢校时）。
• 监控时钟漂移，超阈值告警。
• 关键业务用 fencing token 而不是依赖时钟。

---

11.7 实操：跑一遍配套代码

实战代码见 11_distributed_lock/code/：

• 01_lock_evolution.py：V1 / V2 / V3 三版锁实现，每版复现对应的缺陷
• 02_safe_lock.py：完整的 SET NX EX + Lua 释放 + 唯一 token 实现，并提供上下文管理器
• 03_watchdog.py：用 threading 模拟看门狗，演示「无续期 → 误删」与「有续期 → 安全」对比
• 04_redlock.py：自实现简化版 Redlock，向多个 Redis 实例（或失败时退化为单实例多 DB）请求锁

浏览器演示见 11_distributed_lock/demo.html：

• ① 锁演进史时序图：V1/V2/V3 三种实现对比抢锁动画
• ② 看门狗续期可视化：业务 25s × TTL 10s 场景下，无续期 vs 有看门狗的差异
• ③ Redlock 多实例加锁动画：5 个 Redis 实例的多数派判定 + 总耗时校验

启动方式：

# 准备一个本地 Redis
redis-server &

# 跑代码
cd 11_distributed_lock/code/
python 01_lock_evolution.py
python 02_safe_lock.py
python 03_watchdog.py
python 04_redlock.py

---

11.8 本章小结

┌────────────────────────────────────────────────────────────┐
│                       本章核心要点                            │
├────────────────────────────────────────────────────────────┤
│                                                              │
│  ① 多机环境下 synchronized 失效 → 必须用共享存储              │
│                                                              │
│  ② Redis 单机锁演进：                                          │
│       V1 SETNX+DEL    → 崩溃死锁                             │
│       V2 SETNX+EXPIRE → 两步非原子，仍死锁                    │
│       V3 SET NX EX    → 一条命令搞定 ✅                       │
│                                                              │
│  ③ 5 大补丁：                                                 │
│       1. 必有 TTL                                            │
│       2. 释放校验持有者（Lua 原子）                            │
│       3. 看门狗续期（业务可能超时）                            │
│       4. 可重入（Hash + 计数）                                │
│       5. 公平性（FIFO 排队）                                  │
│                                                              │
│  ④ Redlock：N 个独立实例 + 多数派 + 总耗时校验                │
│       适合「比单实例略强一致」的场景                            │
│       但仍不能解决 GC 暂停 / 时钟回拨问题                      │
│                                                              │
│  ⑤ Kleppmann vs antirez：                                    │
│       绝对正确性 → fencing token + 业务幂等                   │
│       Redlock 只是 effort 上限，不是「100% 互斥」              │
│                                                              │
│  ⑥ 选型：                                                     │
│       一般业务 → 单 Redis + Redisson（V3 + 看门狗）            │
│       强一致   → ZK / etcd + fencing                          │
│       金融     → 数据库唯一约束 + 业务幂等才是终极兜底           │
│                                                              │
└────────────────────────────────────────────────────────────┘

---

11.9 面试高频题

Q1：怎么用 Redis 实现分布式锁？给出完整命令

考察点：基础姿势 + 边界细节。

标准答案：

# 1. 加锁（一条原子命令，必须带 NX 和 EX/PX）
SET resource_name <unique_uuid> NX EX 30

# 2. 释放锁（用 Lua 保证「校验 + 删除」原子）
EVAL "
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
" 1 resource_name <unique_uuid>

关键点：

1. unique_uuid 必须每次加锁不同（推荐 UUID 或「机器+线程+随机数」），用于校验持有者。
2. NX 保证只有不存在时才设置（互斥语义）。
3. EX 30 在同一条命令里完成，避免 SETNX + EXPIRE 的两步非原子问题。
4. 释放锁不能直接 DEL，否则可能误删别人的锁。

加分项：提到 Redisson 的看门狗自动续期、可重入实现、Redlock 算法。

---

Q2：为什么释放锁要用 Lua 脚本？

考察点：原子性理解。

标准答案：

释放锁需要两步：

1. GET key 校验持有者是不是自己；
2. 如果是，则 DEL key。

如果分两条命令发送，两步之间可能发生：

T0  ClientA: GET lock → "A_token"  (确认是自己)
T1  锁恰好在此刻 TTL 到期，Redis 自动删除
T2  ClientB: SET lock B_token NX → 成功，B 拿到锁
T3  ClientA: DEL lock → 把 B 的锁删了！

用 Lua 把 GET + DEL 包成一段脚本，Redis 单线程模型保证脚本内所有命令连续执行不被打断，从而避免上述竞态。

加分项：提到 Lua 在 Redis 7 后被 Functions 扩展，但分布式锁场景仍以 EVAL/EVALSHA 为主。

---

Q3：锁的「续命」问题是什么？怎么解决？

考察点：生产细节 + Redisson 原理。

标准答案：

问题描述：

• 加锁时设了 30s 过期；
• 业务由于慢 SQL / GC / 第三方接口慢，跑了 40s；
• 锁在 30s 时被 Redis 自动释放；
• 客户端 B 拿到锁，进入临界区；
• 此时 A 和 B 同时认为自己持锁 → 数据不一致。

解决方案：看门狗（Watchdog）守护线程

实现要点：

1. 加锁成功后启动一个守护线程；
2. 每隔 TTL / 3 时间检查一次锁是否还在自己手上（用 Lua 校验 + PEXPIRE 续期）；
3. 业务释放锁时通知看门狗停止；
4. 业务客户端崩溃 → 看门狗也死 → 不再续期 → 锁正常 TTL 过期。

Redisson 的实现：默认 30s 过期、每 10s 续期，业务无感知。

加分项：

• 续期前必须校验是不是自己的锁，否则会续走别人的锁。
• 看门狗 + fencing token 配合才是完整方案（避免 GC 暂停期间的 split-brain）。

---

Q4：Redlock 是什么？为什么有争议？

考察点：理论深度 + 工程权衡。

标准答案：

Redlock 是什么：

• antirez 提出的多实例 Redis 分布式锁算法；
• 部署 N 个完全独立（不是主从）的 Redis 实例（典型 N=5）；
• 加锁时依次向所有实例 SET NX EX，只有在 N/2+1 个实例上成功且总耗时小于 TTL才算加锁成功；
• 实际锁的有效时间 = 设定 TTL - 申请耗时；
• 失败时向所有 N 个实例（包括没成功的）发释放命令。

争议（Martin Kleppmann vs antirez）：

Kleppmann 认为 Redlock 在以下场景仍不安全：

1. GC 暂停：客户端 STW 几十秒，锁被别人抢走，恢复后还以为自己持锁；
2. 时钟漂移：NTP 跳变导致多实例「同时认为锁过期」，多数派被破坏；
3. 网络分区：被隔离的客户端不知道锁已经被夺走。

Kleppmann 的建议：任何基于 TTL 的锁都不能保证互斥，必须用 fencing token + 存储侧拒绝旧 token 的写。

antirez 反驳：

1. Redlock 是 AP 系统下 effort 的上限，业务要绝对正确应该结合 fencing token，不能怪算法。
2. GC 暂停问题对 ZK 的 session 也存在，并非 Redlock 独有。
3. 现代 NTP slewing + 监控可控时钟漂移问题。

加分项：给出选型建议——一般业务用单 Redis + Redisson 就够；真正不能错的场景靠业务侧的唯一约束、CAS、幂等表兜底。

---

Q5：主从复制的 Redis 实现锁有什么风险？

考察点：异步复制的副作用。

标准答案：

风险点：Redis 主从复制是异步的，存在以下时序：

T0  ClientA: SET lock A NX EX 30 → master OK
T1  master 把命令通过 backlog 异步发往 slave
T2  master 突然挂掉，命令还没到 slave
T3  Sentinel 把 slave 升为新 master
T4  新 master 上没有 lock 这个 key
T5  ClientB: SET lock B NX EX 30 → 新 master OK
T6  💥 A 和 B 同时持有锁

根本原因：CAP 中 Redis 主从选择了 AP（高可用、弱一致），不保证复制完成才返回 ack。

缓解方案（按代价从低到高）：

1. WAIT 命令：SET ...; WAIT 1 100 强制等 1 个 slave ack 100ms 内完成才返回。但仍不是同步复制，slave ack 后 master 也可能挂。
2. 配置 min-replicas-to-write 1：当少于 1 个 slave 在线时拒绝写。降低主切换风险。
3. Redlock 多独立实例：避免单点。
4. 业务侧幂等兜底：唯一索引、CAS。
5. 换 Zookeeper / etcd：CP 系统天然支持强一致互斥。

加分项：提到「分布式锁不是用来代替业务正确性的，而是性能优化」。

---

Q6：分布式锁 Redis 和 Zookeeper 怎么选？

考察点：CAP + 工程取舍。

标准答案：

维度	Redis	Zookeeper
CAP	AP（高可用，可能丢锁）	CP（强一致，可能不可用）
性能	10W+ QPS，1ms 加锁	1W QPS，10~50ms 加锁
客户端检测	TTL 粗粒度	心跳秒级
公平性	默认无	临时顺序节点天然 FIFO
复杂度	一行命令 / Redisson 开箱	需引入 ZK + Curator
运维	Redis 团队大概率已经有	ZK 通常需要单独部署

选型建议：

高并发、可容忍极端边界（毫秒级双写）：选 Redis
   场景：限流、秒杀去重、定时任务防并发、避免重复发邮件

强一致、不允许任何双主（即使一秒）：选 Zookeeper
   场景：分布式选主、配置发布、金融级业务

不确定：默认 Redis + 业务侧幂等兜底

加分项：

1. ZK 的 watcher 通知机制比 Redis 的轮询/订阅更高效。
2. ZK 的客户端会话断开自动释放锁，比 Redis 的 TTL 精确。
3. 但 ZK 写操作要走 ZAB 多数派 → 性能瓶颈在写。
4. 真正关键的业务（钱），无论选哪个都应该有业务层兜底（唯一索引、对账、最终一致补偿）。

---

📌 下一章预告：第 12 章我们讲缓存设计的「三大经典问题」——穿透（查不存在的数据）、击穿（热点 Key 突然过期）、雪崩（大批 Key 同时过期），以及缓存与数据库的一致性方案（Cache Aside / Read Through / Write Behind）。每一个都是大厂面试的「必送分题」。

🎬 可视化演示

演示加载缓慢或样式异常？点此在新标签页打开 ↗

💻 示例代码

01_lock_evolution.py

"""
Ch11 配套代码 1 / 4 —— 分布式锁演进史 V1 / V2 / V3

本脚本演示三个版本锁实现的优缺点：
  V1: SETNX + DEL          —— 客户端崩溃 → 永久死锁
  V2: SETNX + EXPIRE 两步  —— 两步之间崩溃 → 仍然死锁
  V3: SET key val NX EX    —— 一条原子命令，正确实现 ✅

每个版本都会在 Redis 上真实模拟「持锁线程崩溃」的场景，
并打印其他线程能否在合理时间内拿到锁。
"""

import time
import threading
import uuid

try:
    import redis
except ImportError:
    print("请先 pip install redis")
    raise SystemExit(1)

POOL = redis.ConnectionPool(host="127.0.0.1", port=6379, decode_responses=True)


def section(title: str) -> None:
    print("\n" + "=" * 60)
    print(title)
    print("=" * 60)


def reset(key: str) -> None:
    redis.Redis(connection_pool=POOL).delete(key)


# -------------------------------------------------------------
# V1: SETNX + DEL  —— 致命缺陷：客户端崩了锁永远不释放
# -------------------------------------------------------------
def v1_acquire(r, key: str) -> bool:
    return r.setnx(key, "1") == 1


def v1_release(r, key: str) -> None:
    r.delete(key)


def demo_v1() -> None:
    section("V1: SETNX + DEL  →  崩溃后死锁")
    key = "lock:v1"
    reset(key)
    r = redis.Redis(connection_pool=POOL)

    print("[A] 加锁")
    assert v1_acquire(r, key), "应当加锁成功"
    print("[A] 业务执行中... 模拟进程崩溃（不调 release）")

    # 模拟 A 崩溃，不释放锁

    print("[B] 尝试加锁（最多 3 秒）")
    start = time.time()
    while time.time() - start < 3:
        if v1_acquire(r, key):
            print("[B] 拿到锁")
            return
        time.sleep(0.2)
    print(f"[B] ❌ 等了 {time.time()-start:.1f}s 仍拿不到 —— 死锁演示成功")
    print("    （需要人工 DEL，否则永久占用）")
    reset(key)


# -------------------------------------------------------------
# V2: SETNX + EXPIRE 分两步 —— 两步之间崩仍死锁
# -------------------------------------------------------------
def v2_acquire(r, key: str, ttl: int, *, simulate_crash_between: bool = False) -> bool:
    if r.setnx(key, "1") != 1:
        return False
    if simulate_crash_between:
        print("[A] 💥 SETNX 之后、EXPIRE 之前崩溃")
        raise SystemExit  # 模拟进程死亡
    r.expire(key, ttl)
    return True


def v2_release(r, key: str) -> None:
    r.delete(key)


def demo_v2() -> None:
    section("V2: SETNX + EXPIRE 两步  →  两步之间崩溃仍死锁")
    key = "lock:v2"
    reset(key)
    r = redis.Redis(connection_pool=POOL)

    def crashed_acquire():
        try:
            v2_acquire(r, key, ttl=5, simulate_crash_between=True)
        except SystemExit:
            pass  # 子线程"崩溃"

    t = threading.Thread(target=crashed_acquire)
    t.start(); t.join()

    ttl = r.ttl(key)
    print(f"[A] 崩溃后，锁的 TTL = {ttl}  （-1 = 永不过期，死锁！）")

    print("[B] 尝试加锁（最多 3 秒）")
    start = time.time()
    while time.time() - start < 3:
        if v2_acquire(r, key, ttl=5):
            print("[B] 拿到锁")
            return
        time.sleep(0.2)
    print(f"[B] ❌ 等了 {time.time()-start:.1f}s 仍拿不到 —— 仍然死锁")
    reset(key)


# -------------------------------------------------------------
# V3: SET key val NX EX  —— 原子，推荐 ✅
# -------------------------------------------------------------
def v3_acquire(r, key: str, token: str, ttl: int) -> bool:
    return r.set(key, token, nx=True, ex=ttl) is True


def v3_release(r, key: str, token: str) -> int:
    """ 释放前校验 token，避免误删别人的锁；用 Lua 保证原子。"""
    lua = """
    if redis.call('GET', KEYS[1]) == ARGV[1] then
        return redis.call('DEL', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(lua, 1, key, token)


def demo_v3() -> None:
    section("V3: SET key val NX EX  →  正确实现 ✅")
    key = "lock:v3"
    reset(key)
    r = redis.Redis(connection_pool=POOL)
    token = uuid.uuid4().hex

    print(f"[A] 加锁 (token={token[:8]}, TTL=2s)")
    assert v3_acquire(r, key, token, ttl=2)
    print("[A] 模拟业务后崩溃（不释放）")
    # 不调 release —— 模拟进程死亡

    print("[B] 等待锁过期...")
    start = time.time()
    while time.time() - start < 5:
        b_token = uuid.uuid4().hex
        if v3_acquire(r, key, b_token, ttl=2):
            print(f"[B] ✅ {time.time()-start:.1f}s 后拿到锁，自动恢复！")
            v3_release(r, key, b_token)
            return
        time.sleep(0.3)
    print("[B] 异常：超时未拿到")


if __name__ == "__main__":
    try:
        redis.Redis(connection_pool=POOL).ping()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败 ({e})，请确保 127.0.0.1:6379 可用")
        raise SystemExit(1)

    demo_v1()
    demo_v2()
    demo_v3()
    print("\n✅ 三版对比完成。结论：永远使用 V3（SET ... NX EX ...）")

02_safe_lock.py

"""
Ch11 配套代码 2 / 4 —— 完整安全锁实现

特性：
  ① SET key uuid NX PX  原子加锁
  ② Lua 脚本「校验 + 删除」原子释放
  ③ Lua 脚本「校验 + PEXPIRE」原子续期
  ④ 阻塞 / 非阻塞两种获取语义
  ⑤ Python 上下文管理器，with 块自动释放

并发测试：30 个线程对同一个临界区累加 200 次，
正确实现 → 最终值 = 30 × 200 = 6000；若锁错误，会出现「丢失更新」。
"""

import time
import uuid
import threading
import contextlib

try:
    import redis
except ImportError:
    print("请先 pip install redis"); raise SystemExit(1)


RELEASE_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else
    return 0
end
"""

RENEW_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('PEXPIRE', KEYS[1], ARGV[2])
else
    return 0
end
"""


class RedisLock:
    """ 安全的 Redis 分布式锁（V3 + Lua 释放）。"""

    def __init__(self, client: "redis.Redis", key: str, ttl_ms: int = 30_000):
        self.client = client
        self.key = key
        self.ttl_ms = ttl_ms
        self.token = uuid.uuid4().hex
        self._release_sha = None
        self._renew_sha = None

    def _load_scripts(self):
        if self._release_sha is None:
            self._release_sha = self.client.script_load(RELEASE_LUA)
        if self._renew_sha is None:
            self._renew_sha = self.client.script_load(RENEW_LUA)

    def acquire(self, blocking: bool = True, timeout: float = 10.0,
                retry_interval: float = 0.05) -> bool:
        """ 获取锁。blocking=True 时最多重试 timeout 秒。"""
        self._load_scripts()
        deadline = time.time() + timeout
        while True:
            ok = self.client.set(self.key, self.token, nx=True, px=self.ttl_ms)
            if ok:
                return True
            if not blocking or time.time() >= deadline:
                return False
            time.sleep(retry_interval)

    def release(self) -> bool:
        """ 释放锁（仅释放自己的）。"""
        self._load_scripts()
        return bool(self.client.evalsha(self._release_sha, 1, self.key, self.token))

    def renew(self, ttl_ms: int = None) -> bool:
        """ 续期，扩展过期时间到 ttl_ms。"""
        self._load_scripts()
        ttl = ttl_ms or self.ttl_ms
        return bool(self.client.evalsha(self._renew_sha, 1, self.key, self.token, ttl))

    @contextlib.contextmanager
    def hold(self, **kwargs):
        if not self.acquire(**kwargs):
            raise TimeoutError(f"无法获取锁 {self.key}")
        try:
            yield self
        finally:
            self.release()


# -------------------------------------------------------------
# 并发自测：30 线程争抢同一锁，每个累加共享变量 200 次
# 期望最终值 = 6000；若锁失效会出现丢失。
# -------------------------------------------------------------
def demo_concurrent_counter() -> None:
    print("\n" + "=" * 60)
    print("并发测试：30 线程 × 200 次累加（共享 dict + 锁保护）")
    print("=" * 60)

    pool = redis.ConnectionPool(host="127.0.0.1", port=6379, decode_responses=True)
    counter_key = "demo:counter"
    lock_key = "demo:lock"

    r = redis.Redis(connection_pool=pool)
    r.set(counter_key, 0)
    r.delete(lock_key)

    THREADS, PER = 30, 200

    def worker():
        rr = redis.Redis(connection_pool=pool)
        lock = RedisLock(rr, lock_key, ttl_ms=2000)
        for _ in range(PER):
            with lock.hold(timeout=10):
                # 临界区：读 → 改 → 写（如果没锁会丢更新）
                cur = int(rr.get(counter_key))
                rr.set(counter_key, cur + 1)

    ts = [threading.Thread(target=worker) for _ in range(THREADS)]
    t0 = time.time()
    for t in ts: t.start()
    for t in ts: t.join()
    elapsed = time.time() - t0

    final = int(r.get(counter_key))
    expected = THREADS * PER
    print(f"  期望 = {expected}")
    print(f"  实际 = {final}")
    print(f"  耗时 = {elapsed:.2f}s")
    print("  ✅ 锁正确" if final == expected else "  ❌ 出现丢失")
    r.delete(counter_key, lock_key)


# -------------------------------------------------------------
# 校验 release 不会误删别人的锁
# -------------------------------------------------------------
def demo_release_safety() -> None:
    print("\n" + "=" * 60)
    print("安全校验：A 持锁过期 → B 拿锁 → A 调 release，不能误删 B 的锁")
    print("=" * 60)
    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
    key = "demo:safety"
    r.delete(key)

    a = RedisLock(r, key, ttl_ms=500)  # 0.5s 短 TTL
    b = RedisLock(r, key, ttl_ms=5000)

    assert a.acquire(blocking=False)
    print(f"[A] 加锁 token={a.token[:8]}, TTL=500ms")
    time.sleep(0.7)  # 等过期
    assert b.acquire(blocking=False)
    print(f"[B] 锁已过期，B 重新拿到 token={b.token[:8]}")

    deleted_by_a = a.release()
    print(f"[A] 业务跑完，调 release → 删除? {deleted_by_a}")
    print(f"    （应当为 False —— A 没有误删 B 的锁）")
    assert not deleted_by_a, "❌ A 误删了 B 的锁！"

    still_b_holds = (r.get(key) == b.token)
    print(f"    B 仍持有? {still_b_holds}  ✅")
    b.release()


if __name__ == "__main__":
    try:
        redis.Redis(host="127.0.0.1", port=6379).ping()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败：{e}")
        raise SystemExit(1)

    demo_release_safety()
    demo_concurrent_counter()

03_watchdog.py

"""
Ch11 配套代码 3 / 4 —— 看门狗续期机制

业务执行 6 秒，但锁 TTL 只有 2 秒。
对比两种实现：
  ① 朴素锁：业务跑完锁早就被别人抢走
  ② 看门狗锁：守护线程每 0.6s 自动续期，业务安全完成

看门狗实现要点：
  - 后台 daemon 线程，定期 PEXPIRE
  - 续期前用 Lua 校验持有者
  - 主线程 release 时 Event.set() 通知看门狗退出
"""

import time
import uuid
import threading

try:
    import redis
except ImportError:
    print("请先 pip install redis"); raise SystemExit(1)


RELEASE_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else return 0 end
"""

RENEW_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('PEXPIRE', KEYS[1], ARGV[2])
else return 0 end
"""


def now() -> str:
    return time.strftime('%H:%M:%S', time.localtime()) + f".{int(time.time()*1000)%1000:03d}"


class WatchdogLock:
    """ 带看门狗的安全锁。"""

    def __init__(self, client: "redis.Redis", key: str, ttl_ms: int = 2000):
        self.client = client
        self.key = key
        self.ttl_ms = ttl_ms
        self.token = uuid.uuid4().hex
        self._stop = None
        self._thread = None

    def acquire(self, watchdog: bool = True) -> bool:
        ok = self.client.set(self.key, self.token, nx=True, px=self.ttl_ms)
        if not ok:
            return False
        if watchdog:
            self._start_watchdog()
        return True

    def release(self) -> None:
        self._stop_watchdog()
        self.client.eval(RELEASE_LUA, 1, self.key, self.token)

    def _start_watchdog(self) -> None:
        self._stop = threading.Event()
        interval = self.ttl_ms / 3 / 1000.0
        client = self.client
        key, token, ttl = self.key, self.token, self.ttl_ms

        def loop():
            while not self._stop.wait(interval):
                ok = client.eval(RENEW_LUA, 1, key, token, ttl)
                if ok == 1:
                    print(f"  [{now()}] 🐶 看门狗续期 PEXPIRE {ttl}ms ✓")
                else:
                    print(f"  [{now()}] 🐶 锁已不在自己手上，看门狗退出")
                    return

        self._thread = threading.Thread(target=loop, daemon=True)
        self._thread.start()

    def _stop_watchdog(self) -> None:
        if self._stop is not None:
            self._stop.set()
            if self._thread is not None:
                self._thread.join(timeout=1)


# -------------------------------------------------------------
# 演示一：无看门狗 → 业务超时被抢
# -------------------------------------------------------------
def demo_no_watchdog() -> None:
    print("\n" + "=" * 60)
    print("演示 ①：无看门狗（业务 6s × TTL 2s → 中途被抢）")
    print("=" * 60)
    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
    key = "demo:wd:no"
    r.delete(key)

    a = WatchdogLock(r, key, ttl_ms=2000)
    assert a.acquire(watchdog=False)
    print(f"  [{now()}] [A] 拿到锁，TTL=2s，业务开始（要跑 6s）")

    def thief():
        time.sleep(2.5)  # 等 A 的锁过期
        b = WatchdogLock(r, key, ttl_ms=2000)
        if b.acquire(watchdog=False):
            print(f"  [{now()}] [B] ⚠️ 趁 A 锁过期，把锁抢走了！")
            time.sleep(0.5)
            b.release()

    t = threading.Thread(target=thief)
    t.start()

    for i in range(6):
        time.sleep(1)
        owner = r.get(key)
        flag = "✓ 还是 A" if owner == a.token else ("✗ 已被 B 抢" if owner else "已过期")
        print(f"  [{now()}] [A] 业务进行中 {i+1}/6s … 锁所有者: {flag}")
    t.join()
    a.release()
    print(f"  [{now()}] [A] 业务完成 ❌ —— 中途锁被夺走，临界区被破坏")
    r.delete(key)


# -------------------------------------------------------------
# 演示二：有看门狗 → 安全完成
# -------------------------------------------------------------
def demo_with_watchdog() -> None:
    print("\n" + "=" * 60)
    print("演示 ②：有看门狗（业务 6s × TTL 2s × 续期间隔 ~0.67s）")
    print("=" * 60)
    r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
    key = "demo:wd:yes"
    r.delete(key)

    a = WatchdogLock(r, key, ttl_ms=2000)
    assert a.acquire(watchdog=True)
    print(f"  [{now()}] [A] 拿到锁，TTL=2s，业务开始（要跑 6s）")

    def thief():
        for i in range(6):
            time.sleep(1)
            b = WatchdogLock(r, key, ttl_ms=2000)
            got = b.acquire(watchdog=False)
            if got:
                print(f"  [{now()}] [B] ⚠️ 抢锁成功（不该发生！）")
                b.release()

    t = threading.Thread(target=thief)
    t.start()

    for i in range(6):
        time.sleep(1)
        owner = r.get(key)
        ok = "✓ A 持锁" if owner == a.token else "✗ 锁丢了"
        print(f"  [{now()}] [A] 业务 {i+1}/6s … {ok}")

    t.join()
    a.release()
    print(f"  [{now()}] [A] 业务完成 ✅ —— 看门狗保住了临界区")
    r.delete(key)


if __name__ == "__main__":
    try:
        redis.Redis(host="127.0.0.1", port=6379).ping()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败：{e}")
        raise SystemExit(1)

    demo_no_watchdog()
    demo_with_watchdog()

04_redlock.py

"""
Ch11 配套代码 4 / 4 —— 自实现简化版 Redlock

完整算法步骤（见 11.4.3）：
  Step 1: 记录 t1
  Step 2: 依次向 N 个实例 SET ... NX PX，单实例超时极小
  Step 3: 记录 t2，计算 Δt = t2 - t1
  Step 4: 多数派成功 (≥N/2+1) 且 Δt < TTL 才算加锁成功
  Step 5: 实际有效时间 = TTL - Δt
  Step 6: 失败时也要去所有实例释放（防止网络抖动导致部分成功）

环境优雅降级：
  如果你只有一个 Redis 实例，本脚本会自动用同一实例的 5 个不同 DB
  来模拟「5 个独立实例」（仅用于演示算法流程，不具备 Redlock 的隔离性）。
"""

import time
import uuid
from typing import List, Tuple

try:
    import redis
except ImportError:
    print("请先 pip install redis"); raise SystemExit(1)


RELEASE_LUA = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
else return 0 end
"""

CLOCK_DRIFT_FACTOR = 0.01  # antirez 论文给的经验值（1%）
SINGLE_REQUEST_TIMEOUT = 0.05  # 50ms，单实例阻塞上限


class RedLock:
    def __init__(self, instances: List["redis.Redis"], key: str, ttl_ms: int = 5000):
        self.instances = instances
        self.key = key
        self.ttl_ms = ttl_ms
        self.token = uuid.uuid4().hex
        self.quorum = len(instances) // 2 + 1
        self.valid_ms = 0  # 加锁成功后的实际有效时间

    def _try_acquire_one(self, client) -> bool:
        try:
            return client.set(self.key, self.token, nx=True, px=self.ttl_ms) is True
        except Exception:
            return False

    def _release_one(self, client) -> None:
        try:
            client.eval(RELEASE_LUA, 1, self.key, self.token)
        except Exception:
            pass

    def acquire(self) -> Tuple[bool, dict]:
        """ 尝试加锁。返回 (是否成功, 详细信息)。"""
        t1 = time.time()

        votes = []  # [(idx, ok)]
        for idx, c in enumerate(self.instances):
            ok = self._try_acquire_one(c)
            votes.append((idx, ok))

        elapsed_ms = int((time.time() - t1) * 1000)
        drift_ms = int(self.ttl_ms * CLOCK_DRIFT_FACTOR) + 2
        valid_ms = self.ttl_ms - elapsed_ms - drift_ms
        success_count = sum(1 for _, ok in votes if ok)

        info = {
            "votes": votes,
            "success_count": success_count,
            "quorum": self.quorum,
            "elapsed_ms": elapsed_ms,
            "valid_ms": valid_ms,
        }

        if success_count >= self.quorum and valid_ms > 0:
            self.valid_ms = valid_ms
            return True, info

        # 失败：去所有实例释放（包括没成功的，防止半成功）
        for c in self.instances:
            self._release_one(c)
        return False, info

    def release(self) -> None:
        for c in self.instances:
            self._release_one(c)


def build_instances() -> List["redis.Redis"]:
    """ 优先尝试 5 个端口；若不可用则降级为 1 个端口 + 5 个 DB。"""
    candidates = [
        ("127.0.0.1", 6379), ("127.0.0.1", 6380), ("127.0.0.1", 6381),
        ("127.0.0.1", 6382), ("127.0.0.1", 6383),
    ]
    inst = []
    for host, port in candidates:
        try:
            c = redis.Redis(host=host, port=port, socket_timeout=SINGLE_REQUEST_TIMEOUT,
                            socket_connect_timeout=0.2, decode_responses=True)
            c.ping()
            inst.append(c)
        except Exception:
            pass

    if len(inst) >= 3:
        print(f"✅ 检测到 {len(inst)} 个独立 Redis 实例，启用真 Redlock")
        return inst

    print("⚠️  未检测到多实例，降级使用单实例 × 5 DB（仅演示算法流程）")
    return [redis.Redis(host="127.0.0.1", port=6379, db=i, decode_responses=True,
                        socket_timeout=SINGLE_REQUEST_TIMEOUT) for i in range(5)]


# -------------------------------------------------------------
# 演示 1: 正常 5 实例加锁成功
# -------------------------------------------------------------
def demo_normal_acquire() -> None:
    print("\n" + "=" * 60)
    print("演示 ①：正常加锁（期望多数派 + 总耗时 < TTL）")
    print("=" * 60)
    instances = build_instances()
    lock = RedLock(instances, key="redlock:order:1001", ttl_ms=5000)
    ok, info = lock.acquire()
    print(f"  实例投票: {info['votes']}")
    print(f"  成功数: {info['success_count']} / 多数派需 {info['quorum']}")
    print(f"  申请耗时: {info['elapsed_ms']} ms")
    print(f"  实际有效时间: {info['valid_ms']} ms")
    print(f"  结果: {'✅ 加锁成功' if ok else '❌ 加锁失败'}")
    if ok:
        lock.release()


# -------------------------------------------------------------
# 演示 2: 模拟 2 个实例已被占 → 仍能多数派成功
# -------------------------------------------------------------
def demo_partial_failure() -> None:
    print("\n" + "=" * 60)
    print("演示 ②：模拟 2 个实例被预先占用（5 个里仍能拿到 3 个，多数派成功）")
    print("=" * 60)
    instances = build_instances()
    key = "redlock:order:1002"
    # 预占 2 个
    for c in instances[:2]:
        c.set(key, "intruder", px=10_000)
    lock = RedLock(instances, key=key, ttl_ms=5000)
    ok, info = lock.acquire()
    print(f"  实例投票: {info['votes']}  （前 2 个被占用）")
    print(f"  成功数: {info['success_count']} / 多数派需 {info['quorum']}")
    print(f"  结果: {'✅ 加锁成功' if ok else '❌ 加锁失败'}")
    if ok:
        lock.release()
    # 清理
    for c in instances[:2]:
        c.delete(key)


# -------------------------------------------------------------
# 演示 3: 模拟 3 个实例被占 → 不足多数派，加锁失败
# -------------------------------------------------------------
def demo_quorum_fail() -> None:
    print("\n" + "=" * 60)
    print("演示 ③：3 个实例被占 → 仅 2 个可加锁 → 不足多数派 → 失败")
    print("=" * 60)
    instances = build_instances()
    key = "redlock:order:1003"
    for c in instances[:3]:
        c.set(key, "intruder", px=10_000)
    lock = RedLock(instances, key=key, ttl_ms=5000)
    ok, info = lock.acquire()
    print(f"  实例投票: {info['votes']}  （前 3 个被占用）")
    print(f"  成功数: {info['success_count']} / 多数派需 {info['quorum']}")
    print(f"  结果: {'✅ 加锁成功' if ok else '❌ 加锁失败（符合预期）'}")
    for c in instances[:3]:
        c.delete(key)


if __name__ == "__main__":
    try:
        demo_normal_acquire()
        demo_partial_failure()
        demo_quorum_fail()
    except redis.ConnectionError as e:
        print(f"❌ Redis 连接失败：{e}")
        raise SystemExit(1)

    print("\n说明：本实现是 Redlock 算法的简化教学版。")
    print("生产环境推荐 Java 用 Redisson 的 RedissonRedLock，")
    print("Python 可参考 `redlock-py` 库（pip install redlock-py）。")

01_lock_evolution.py ↗ · 02_safe_lock.py ↗ · 03_watchdog.py ↗ · 04_redlock.py ↗