Redis 分布式锁从SETNX 到 Redlock 的问题记录_Redis

分布式锁看起来简单——setnx 加锁，del 释放。
但真到生产环境，你要处理：进程崩溃锁不释放、
删了别人的锁、业务没执行完锁就过期、
主从切换导致锁失效……

本篇把这些坑一个个列出来，给出能用的解法，
最后说清楚 redlock 到底该不该用。

一、分布式锁的最低要求

一个可用的分布式锁至少要满足：

互斥性：任意时刻只有一个客户端能持有
不会死锁：持有者崩溃后锁能自动释放
加锁和解锁是同一个客户端（不能删别人的锁）
容错性：部分 redis 节点宕机时仍能工作

二、演进：五个版本的写法

v1：setnx + del（❌ 会死锁）

r.setnx("lock", 1)
try:
    do_something()
finally:
    r.delete("lock")

问题：如果 do_something() 时进程崩溃，
finally 不会执行，锁永远不释放，死锁。

v2：setnx + expire（❌ 非原子）

r.setnx("lock", 1)
r.expire("lock", 30)      # 加过期时间

问题：这两条命令不是原子的。
如果 setnx 成功后进程崩溃，还没来得及 expire，
又是死锁。

v3：set nx ex（✅ 可用，但仍有问题）

r.set("lock", token, nx=true, ex=30)

redis 2.6.12+ 支持 set key value nx ex seconds，
加锁和设过期时间是原子的。这是正确的基础写法。

但仍有问题——会删掉别人的锁：

# 客户端 a 加锁，超时 30 秒
# a 的业务执行了 35 秒（超过 30 秒），锁已自动释放
# 客户端 b 拿到锁
# a 执行完，del 删除锁 ← 把 b 的锁删了！

v4：lua 脚本原子释放（✅ 生产可用）

释放锁时先比对 value 再删除，用 lua 保证原子性：

import uuid
import redis
r = redis.redis()
def acquire(lock_key, ttl=30):
    token = str(uuid.uuid4())          # 唯一标识
    ok = r.set(lock_key, token, nx=true, ex=ttl)
    return token if ok else none
def release(lock_key, token):
    # ⚠️ 必须是 lua：get 和 del 要原子执行
    script = """
    if redis.call('get', keys[1]) == argv[1] then
        return redis.call('del', keys[1])
    else
        return 0
    end
    """
    return r.eval(script, 1, lock_key, token)
# 使用
token = acquire("lock:order:123", ttl=30)
if token:
    try:
        do_something()
    finally:
        release("lock:order:123", token)
else:
    print("没拿到锁")

🔑 为什么必须用 lua？

# ❌ 错误：两步操作非原子
if r.get(lock_key) == token:      # 判断时是我的锁
    # 恰好这一刻锁过期了，b 拿到了锁
    r.delete(lock_key)             # 删掉了 b 的锁

get 和 del 之间有时间窗口，必须合并成一个原子操作。

v5：自动续期（看门狗）

问题：业务执行时间不确定，30 秒可能不够。

解法：看门狗（watchdog）——
加锁成功后起一个后台线程，
每隔 ttl/3 时间检查锁还在不在，在就延长。

import threading, time
class redislock:
    def __init__(self, r, key, ttl=30):
        self.r, self.key, self.ttl = r, key, ttl
        self.token = none
        self._timer = none
    def __enter__(self):
        self.token = str(uuid.uuid4())
        if not self.r.set(self.key, self.token, nx=true, ex=self.ttl):
            self.token = none
            raise runtimeerror("获取锁失败")
        self._start_watchdog()
        return self
    def __exit__(self, *exc):
        self._stop_watchdog()
        self._release()
    def _start_watchdog(self):
        """每 ttl/3 秒续期一次"""
        def renew():
            if self.token:
                # 只有锁还是我的才续期
                self.r.eval("""
                    if redis.call('get', keys[1]) == argv[1] then
                        return redis.call('pexpire', keys[1], argv[2])
                    end
                    return 0
                """, 1, self.key, self.token, self.ttl * 1000)
            self._timer = threading.timer(self.ttl / 3, renew)
            self._timer.daemon = true
            self._timer.start()
        self._timer = threading.timer(self.ttl / 3, renew)
        self._timer.daemon = true
        self._timer.start()
    def _stop_watchdog(self):
        if self._timer:
            self._timer.cancel()
            self._timer = none
    def _release(self):
        if self.token:
            self.r.eval("""
                if redis.call('get', keys[1]) == argv[1] then
                    return redis.call('del', keys[1])
                end
                return 0
            """, 1, self.key, self.token)
            self.token = none
# 使用
with redislock(r, "lock:order:123"):
    do_something()      # 执行多久都不怕，看门狗会自动续期

这就是 redisson 的 watchdog 机制的原理。

三、用现成的库：redisson

java 项目直接用 redisson，上面所有功能它都实现好了：

rlock lock = redissonclient.getlock("lock:order:123");
// 自动续期（默认 30 秒 ttl，每 10 秒续一次）
lock.lock();
try {
    dosomething();
} finally {
    lock.unlock();
}
// 或者指定超时时间（不续期）
boolean acquired = lock.trylock(10, 30, timeunit.seconds);

redisson 额外提供的能力：

可重入锁：同一线程可多次加锁（内部用 hash 记重入次数）
公平锁：按请求顺序获取
读写锁：rreadwritelock
联锁 / 红锁

python 对应的是 redis-py 的 lock，但功能弱很多，
没有看门狗。生产建议自己封装或用 redlock-py。

四、主从切换导致的锁失效

这是单 redis 实例方案的根本缺陷：

1. 客户端 a 在 master 上加锁成功
2. master 还没把锁同步给 slave，就宕机了
3. slave 被提升为新 master
4. 客户端 b 在新 master 上加锁 ← 也成功了！
5. a 和 b 同时持有锁 → 互斥性被破坏

解决方案：redlock

redlock 的思路：向 n 个独立的 redis 实例（通常 5 个）依次加锁，
超过半数（n/2 + 1）成功才算加锁成功。

客户端
  ├─→ redis 1  ✓
  ├─→ redis 2  ✓
  ├─→ redis 3  ✓   ← 3/5 成功，加锁成功
  ├─→ redis 4  ✗
  └─→ redis 5  ✗

关键约束：

n 个实例互相独立（不是同一个集群的分片，也不是主从）
加锁有超时时间，超时就跳过这个节点
总耗时必须小于锁的 ttl，否则认为失败
释放时向所有实例发删除请求

from redlock import redlock
dlm = redlock([
    {"host": "redis1", "port": 6379},
    {"host": "redis2", "port": 6379},
    {"host": "redis3", "port": 6379},
    {"host": "redis4", "port": 6379},
    {"host": "redis5", "port": 6379},
])
lock = dlm.lock("lock:order", 1000)      # ttl 1000 毫秒
if lock:
    try:
        do_something()
    finally:
        dlm.unlock(lock)

⚠️ redlock 有争议，谨慎使用

redis 作者 antirez 和分布式系统专家 martin kleppmann
曾就此激烈争论。主要的质疑：

依赖时钟：redlock 假设各节点时钟大致同步。
如果发生时钟跳跃（ntp 调整、运维改时间），
锁的 ttl 计算会出错
gc 停顿：客户端 gc 停顿期间锁可能过期，
但客户端以为自己还持有
复杂度高，收益存疑：为了极小概率的主从切换场景，
引入 5 个实例和可观的延迟

现实建议：

场景	建议
一般业务（防止重复扣款、防止并发写）	单实例 + lua 释放 + 看门狗，够用
对正确性要求极高（金融）	别用 redis 锁，用数据库的乐观锁 / zookeeper / etcd
必须用 redlock	确保有 fencing token（递增序号）做最终兜底

🔑 fencing token 是什么：
每次加锁返回一个单调递增的序号，
操作数据时带上这个序号，存储层拒绝处理序号更小的请求。
这样即使锁失效了（gc 停顿导致），
旧的请求也会被存储层挡住。这才是真正的兜底。

五、分布式锁的替代方案

很多时候根本不需要分布式锁。

5.1 数据库乐观锁（推荐）

update products
set stock = stock - 1, version = version + 1
where id = 1001 and stock >= 1 and version = 5;
-- 影响行数 0 说明被别人改过，重试

用 cas + 版本号，不需要 redis，
且天然和数据库事务一致。扣库存这类场景首选这个。

5.2 数据库悲观锁

begin;
select * from products where id = 1001 for update;   -- 行锁
update products set stock = stock - 1 where id = 1001;
commit;

简单可靠，但要注意锁的范围和事务时长（见 mysql 锁那篇）。

5.3 lua 脚本原子操作

很多时候"锁"要解决的是多个操作的原子性，
那不如直接把多个操作写成一个 lua 脚本：

# 扣库存：判断 + 扣减一步完成，不需要锁
script = """
local stock = tonumber(redis.call('get', keys[1]))
if stock <= 0 then
    return -1
end
redis.call('decr', keys[1])
return stock - 1
"""
r.eval(script, 1, "stock:1001")

lua 脚本在 redis 里是原子执行的，
这是最高效的"锁替代方案"。

选型建议

场景	方案
单纯防止重复执行	数据库唯一索引 / 乐观锁
扣库存、改余额	lua 脚本或 db 乐观锁
需要跨多个资源的原子操作	分布式锁
定时任务只在一台机器跑	分布式锁（或 k8s lease）
金融级强一致	别用 redis，用 etcd / zk

六、八个必须知道的坑

坑 1：忘了设过期时间 → 死锁。用 set nx ex 一步到位。

坑 2：del 之前不比对 token → 删别人的锁。用 lua。

坑 3：ttl 太短，业务没跑完锁就没了 → 用看门狗续期。

坑 4：把锁的粒度设得太大

with lock("global_lock"):      # ❌ 全局锁，所有请求串行
    ...
with lock(f"order:{order_id}"):   # ✅ 按资源加锁
    ...

坑 5：锁重入

def a():
    with lock: b()      # 外层加了锁
def b():
    with lock: ...      # ❌ 不可重入的话这里死锁

需要可重入就用 redisson，或者自己用 hash 记重入次数。

坑 6：锁的 value 用了固定值
必须用 uuid 等唯一标识，否则无法区分持有者。

坑 7：redis 单实例宕机 → 锁服务不可用。
要么接受（多数场景可接受），要么上 redlock，要么换 etcd。

坑 8：主从切换丢锁（前面详述）→ 这是单实例方案的固有缺陷。

七、一份可以直接用的实现

import uuid, threading, time
import redis
class distlock:
    """生产可用的 redis 分布式锁（单实例版）"""
    _release = """
    if redis.call('get', keys[1]) == argv[1] then
        return redis.call('del', keys[1])
    end
    return 0
    """
    _renew = """
    if redis.call('get', keys[1]) == argv[1] then
        return redis.call('pexpire', keys[1], argv[2])
    end
    return 0
    """
    def __init__(self, r, key, ttl=30, auto_renew=true, retry=0, retry_delay=0.1):
        self.r, self.key, self.ttl = r, key, ttl
        self.auto_renew = auto_renew
        self.retry, self.retry_delay = retry, retry_delay
        self.token = none
        self._timer = none
    def acquire(self):
        self.token = str(uuid.uuid4())
        for i in range(self.retry + 1):
            if self.r.set(self.key, self.token, nx=true, ex=self.ttl):
                if self.auto_renew:
                    self._schedule_renew()
                return true
            if i < self.retry:
                time.sleep(self.retry_delay)
        self.token = none
        return false
    def release(self):
        if not self.token:
            return
        if self._timer:
            self._timer.cancel()
            self._timer = none
        self.r.eval(self._release, 1, self.key, self.token)
        self.token = none
    def _schedule_renew(self):
        def renew():
            if self.token:
                try:
                    self.r.eval(self._renew, 1, self.key,
                                self.token, int(self.ttl * 1000))
                except exception:
                    pass
                self._timer = threading.timer(self.ttl / 3, renew)
                self._timer.daemon = true
                self._timer.start()
        self._timer = threading.timer(self.ttl / 3, renew)
        self._timer.daemon = true
        self._timer.start()
    def __enter__(self):
        if not self.acquire():
            raise runtimeerror(f"获取锁失败: {self.key}")
        return self
    def __exit__(self, *exc):
        self.release()
# 使用
with distlock(r, "lock:order:123", ttl=30, retry=3):
    do_something()

八、小结

最低可用写法：set key token nx ex 30 + lua 比对 token 释放
必须加过期时间，且加锁和设 ttl 要原子（nx ex 一起用）
释放锁必须 lua，否则 get 和 del 之间的窗口会删掉别人的锁
业务耗时不确定 → 看门狗自动续期（每 ttl/3 续一次）
单实例方案的主从切换会丢锁，这是固有缺陷
redlock 有争议，谨慎使用；真要强一致请上 etcd / zookeeper
很多场景根本不需要锁：扣库存用 lua 或 db 乐观锁更好
锁的粒度要按资源分，别用全局锁

到此这篇关于redis 分布式锁讲透：从 setnx 到 redlock 的所有坑的文章就介绍到这了,更多相关redis 分布式锁从setnx到redlock内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！

Redis 分布式锁从SETNX 到 Redlock 的问题记录

一、分布式锁的最低要求

二、演进：五个版本的写法

v1：setnx + del（❌ 会死锁）

v2：setnx + expire（❌ 非原子）

v3：set nx ex（✅ 可用，但仍有问题）

v4：lua 脚本原子释放（✅ 生产可用）

v5：自动续期（看门狗）

三、用现成的库：redisson

四、主从切换导致的锁失效

解决方案：redlock

⚠️ redlock 有争议，谨慎使用

五、分布式锁的替代方案

5.1 数据库乐观锁（推荐）

5.2 数据库悲观锁

5.3 lua 脚本原子操作

选型建议

六、八个必须知道的坑

七、一份可以直接用的实现

八、小结

推荐阅读

Redis Pipeline写入数据乱序是什么原因导致的

Redis未授权访问漏洞怎么检测和修复？一文看懂6379端口安全风险

使用Redis来实现锁的多种方式

Redis 多机房同步实现主从跨地域复制延迟与数据一致性

在Ubuntu上安装和使用Composer流程步骤

Nginx+Lua+Redis实现动态封禁IP的几种方法

猜你喜欢

发表评论