《深入理解 Redis 分布式锁：为何必须使用 SET NX EX + Lua 脚本？》

一、引言：分布式系统中的“锁”之痛

在单体应用中，加锁是件简单的事：synchronized、ReentrantLock 足以应对大多数并发场景。但当系统演进为分布式架构，多个服务节点同时访问共享资源时，传统锁机制便失去了效力。

于是，分布式锁应运而生。它的目标是：在多个进程或服务之间协调对共享资源的互斥访问，确保数据一致性与操作原子性。而在众多实现方案中，基于 Redis 的分布式锁因其部署简单、性能优越而被广泛采用。

但你是否曾想过：

• 为什么不能直接用 SETNX 实现锁？
• 为什么推荐使用 SET key value NX EX？
• 为什么释放锁时一定要用 Lua 脚本？
• Redlock 是银弹吗？是否真的“安全”？

本文将带你从原理、实践到最佳实践，逐一解答这些问题。

---

二、Redis 分布式锁的基本实现方式

1. 最朴素的实现：SETNX

# Python 示例：尝试加锁
import redis
r = redis.Redis()

lock_acquired = r.setnx("my_lock", "unique_id")
if lock_acquired:
    # 执行业务逻辑
    ...
    r.delete("my_lock")

问题：

• 若程序在执行业务逻辑前宕机，锁将永远不释放，造成死锁。
• 没有设置过期时间，锁的生命周期无法控制。

2. 改进版：SETNX + EXPIRE

if r.setnx("my_lock", "unique_id"):
    r.expire("my_lock", 10)

问题：

• setnx 与 expire 是两个独立命令，非原子操作。
• 若在 setnx 成功后宕机，expire 未执行，仍可能造成死锁。

---

三、推荐方式：SET key value NX EX

Redis 从 2.6.12 起支持如下命令：

SET key value NX EX 10

含义：

• NX：仅当 key 不存在时才设置（即 SETNX）
• EX 10：设置过期时间为 10 秒
• 原子性：这是一个原子操作！

Python 示例（使用 redis-py）：

r.set("my_lock", "unique_id", nx=True, ex=10)

为什么要设置唯一值？

因为释放锁时需要判断当前客户端是否拥有锁，避免误删他人锁。

# 锁值应为唯一标识，如 UUID
import uuid
lock_value = str(uuid.uuid4())
r.set("my_lock", lock_value, nx=True, ex=10)

---

四、释放锁：为什么必须使用 Lua 脚本？

场景还原：

1. 客户端 A 获取锁，值为 uuid-a
2. 锁即将过期，客户端 B 尝试获取锁
3. 客户端 A 执行业务逻辑结束，调用 DEL my_lock

问题：

• 若锁已过期，B 已获取锁，A 的 DEL 会误删 B 的锁！

正确做法：判断值后再删除

-- Lua 脚本：仅当值匹配时才删除
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end

Python 示例：

unlock_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
"""

unlock = r.register_script(unlock_script)
unlock(keys=["my_lock"], args=[lock_value])

优势：

• 原子性：Lua 脚本在 Redis 中一次性执行，避免竞态条件。
• 安全性：确保只有持有锁的客户端才能释放锁。

---

五、Redlock：分布式锁的“终极方案”？

Redlock 是 Redis 作者提出的一种跨多个 Redis 实例的分布式锁算法，核心思想：

• 在多个 Redis 节点上并行尝试加锁（如 5 个节点）
• 若在指定时间内成功获取多数节点（如 3 个），则认为加锁成功
• 设置统一的过期时间
• 释放锁时逐个释放

Python 实现（使用 redis-py + redis.lock）：

from redis import Redis
from redis.lock import Lock

r = Redis()
lock = Lock(r, "my_lock", timeout=10)
if lock.acquire(blocking=False):
    try:
        # 执行业务逻辑
        ...
    finally:
        lock.release()

Redlock 的争议：

• 网络分区时可能出现“脑裂”
• 多节点时时钟漂移影响判断
• 官方文档也指出：单 Redis 实例 + SET NX EX + Lua 脚本已足够安全

---

六、最佳实践与实战建议

1. 锁的唯一标识

使用 UUID 或线程 ID，确保锁的归属可验证。

import uuid
lock_value = str(uuid.uuid4())

2. 设置合理的过期时间

• 过短：业务未完成锁已失效，导致并发问题
• 过长：宕机后锁迟迟不释放，影响其他客户端

建议：业务预估时间 × 安全系数（如 1.5）

3. 自动续期机制（可选）

若业务执行时间不可预期，可使用后台线程定期续期。

参考实现：redlock-py、python-redis-lock 等库。

4. 封装为上下文管理器

from contextlib import contextmanager

@contextmanager
def redis_lock(r, key, value, timeout=10):
    if r.set(key, value, nx=True, ex=timeout):
        try:
            yield
        finally:
            unlock_script = """
            if redis.call("get", KEYS[1]) == ARGV[1] then
                return redis.call("del", KEYS[1])
            else
                return 0
            end
            """
            r.eval(unlock_script, 1, key, value)
    else:
        raise Exception("Lock acquisition failed")

使用方式：

with redis_lock(r, "my_lock", lock_value):
    # 安全执行任务
    ...

---

七、总结与展望

为什么必须使用 SET NX EX + Lua？

• 原子性：避免竞态条件
• 安全性：防止误删他人锁
• 简洁性：一行命令搞定加锁，Lua 保证释放安全

未来展望：

• Redis 7 引入了 SET 命令的新参数 GET，可进一步优化锁释放逻辑
• Redis 社区持续演进，未来可能支持更强的事务语义
• Python 社区也在不断涌现更强的分布式锁库，如 aioredlock、fastapi-limiter 等

---

八、互动时间：你怎么看？

• 你在实际项目中使用过 Redis 分布式锁吗？遇到过哪些坑？
• 是否尝试过 Redlock？你如何看待它的安全性与复杂度？
• 除了 Redis，你还用过哪些分布式锁方案（如 ZooKeeper、Etcd）？

欢迎在评论区留言交流，我们一起构建更健壮的分布式系统！

---

附录与参考资料

• Redis 官方文档
• SET 命令说明
• Redlock 算法原理 (redis.io in Bing)
• 推荐阅读：
  • 《Redis 实战》
  • 《Python 编程：从入门到实践》
  • 《流畅的 Python》