简介

业务中，常有分布式锁的需求，常见的解决方案便是基于 Redis 作为中心节点实现伪分布式效果，因为存在中心节点，所以我将其定义为伪分布式。

回归主题，这篇文章，主要理一下，基于 Redis 实现简单分布式锁的一些问题，Redis 支持 RedLock(红锁)等复杂的实现，以后的文章再讨论。

基于 SETNX 命令实现分布式锁

使用 SETNX 命令构建分布式锁是最常见的实现方式，具体而言：

1. 通过 SETNX key value 向 Redis 新增一个值，SETNX 命令只有当 key 不存在时，才会插入值并返回成功，否则返回失败，而 KEY 便可以作为分布式锁的锁名，通常基于业务来决定该锁名；

2. 通过 DEL key 命令删除 key，从而实现释放锁的效果，当锁释放后，其他线程才可以通过 SETNX 获得锁(相同的 KEY)；

3. 利用 EXPIRE key timeout 对 KEY 设置超时时间，从而实现锁的超时自动释放的效果，避免资源一直被占用。

redis-py (https://github.com/redis/redis-py) 这个库便基于这种形式实现 Redis 分布式锁，将其源码中相关代码复制出来，如下：

# 获得分布式锁
def do_acquire(self, token): # 利用SETNX实现分布式锁
    if self.redis.setnx(self.name, token): if self.timeout: timeout = int(self.timeout * 1000) # 转成毫秒
            # 设置分布式超时时间
            self.redis.pexpire(self.name, timeout) return True return False # 释放分布式锁
def do_release(self, expected_token): name = self.name def execute_release(pipe): lock_value = pipe.get(name) if lock_value != expected_token: raise LockError("Cannot release a lock that's no longer owned") # 利用DEL value实现锁的释放
        pipe.delete(name) self.redis.transaction(execute_release, name)

这种方式，存在一些问题，下文进行简单的分析。

SETNX 与 EXPIRE 非原子性问题

SETNX 与 EXPIRE 是两个操作，在 Redis 中不是原子操作。

如果 SETNX 成功(即获得锁)，但在通过 EXPIRE 设置锁超时时间时，服务器挂机、网络中断等问题，导致 EXPIRE 没有成功执行，此时锁就变成了没有超时时间的锁了，如果业务逻辑没有处理好锁的释放，则容易出现死锁。

Redis 官方考虑到了这种情况，让 SET 命令可以直接设置 Timeout 并实现 SETNX 效果，SET 支持的语法变为：SETEX key value NX timeout，这样就不再需要通过 EXPIRE 设置超时时间，从而实现原子性了。

当然，在 Redis 官方还没有实现这一功能时，很多开源库也考虑到了这个问题，然后使用 Lua 脚本实现 SETEX 与 EXPIRE 两个操作的原子性。

因为用户希望自定义若干指令来完成特定的业务，Redis 官方为这些用户提供了 Lua 脚本支持，用户可以向 Redis 服务器发送 Lua 脚本执行自定义的逻辑，Redis 服务器会单线程原子性的执行 Lua 脚本。

锁误解除

锁误解除也是常见的情况。

假设现在有 A、B 两个线程在工作并竞争同一把锁，线程 A 获得了锁，并将锁的超时时间设置完成 30s，但线程 A 在处理业务逻辑时，因为数据库 SQL 超时，原本 20s 就可以完成的任务，现在需要 40s 才能完成，当线程 A 花费 30s 时，锁会自动释放，此时线程 B 会获得这把锁，当线程 A 处理完业务逻辑时，会通过 DEL 去释放锁，此时释放的是线程 B 的锁，直观如下图所示：

解决方法便是添加唯一标识，在释放锁时，校验 KEY 对应的唯一标识是否被当前线程持有，在 redis-py 中，通过 UUID 生成了当前线程的唯一标识 token，并在释放锁时，判断当前线程是否拥有相同的 token，相关代码如下 (你会发现与上面复制出来的代码不同，这是因为旧文中使用的 redis-py 版本为 2.10.6，现在使用的 redis-py 版本为 3.5.3，相关的 bug 已经被修改了，旧文的代码，只是为了引出问题)：

class Lock(object): def __init__(self, redis, name, timeout=None, sleep=0.1, blocking=True, blocking_timeout=None, thread_local=True): # 线程本地存储
        self.local = threading.local() if self.thread_local else dummy() self.local.token = None


    def acquire(self, blocking=None, blocking_timeout=None, token=None): sleep = self.sleep if token is None: # 基于UUID算法生成唯一token
            token = uuid.uuid1().hex.encode() # 省略剩余代码...

    def do_acquire(self, token): if self.timeout: timeout = int(self.timeout * 1000) else: timeout = None
        # Token会通过set方法存入到Redis中
        if self.redis.set(self.name, token, nx=True, px=timeout): return True return False

redis-py 基于 uuid 库生成 token，并将其存到当前线程的本地存储空间中(独立于其他线程)，在释放时，判断当前线程的 token 与加锁时存储的 token 释放相同，redis-py 中利用 Lua 来实现这个过程，相关代码如下：

def release(self): "Releases the already acquired lock" # 从线程本地存储中获得token
    expected_token = self.local.token if expected_token is None: raise LockError("Cannot release an unlocked lock") self.local.token = None
    self.do_release(expected_token) def do_release(self, expected_token): # 利用Lua来释放锁，并实现判断token是否相同的逻辑
    if not bool(self.lua_release(keys=[self.name], args=[expected_token], client=self.redis)): raise LockNotOwnedError("Cannot release a lock" " that's no longer owned") 

其中 lua_release 变量具体的值为：

LUA_RELEASE_SCRIPT = """  local token = redis.call('get', KEYS[1])  if not token or token ~= ARGV[1] then  return 0  end  redis.call('del', KEYS[1])  return 1  """ 

上述 Lua 代码中，通过 get 获得 KEY 的 value，这个 value 就是 token，然后判断与传入的 token 是否相同，不相同的话，便不会执行 DEL 命令，即不会释放锁。

锁超时导致的并发

这种情况与锁误解除类似，同样假设有线程 A、B，线程 A 获得锁并设置过期时间 30s，当线程 A 执行时间超过 30s 时，锁过期释放，此时线程 B 获得锁，如果线程 A 与线程 B 是在业务上是有顺序依赖的，此时出现了并发情况，便会导致业务结果的错误，直观如下图：

线程 A、B 同时执行导致业务错误是我们不希望出现的，对于这种情况，有两种解决方案：

1. 增大锁的过期时间，让业务逻辑有充足的执行时间；

2. 添加守护线程，当锁过期时，添加过期时间。

建议使用第一种方案，简单直接，此外，可以添加单一线程，对 Redis 的 key 做监控，对于时长特别长的 key，做监控报警。

轮询等待的效率问题

依旧是线程 A、B，当线程 A 获得锁时，线程 B 也想获得锁，此时就需要等待，直到线程 A 释放锁或者锁过期自己释放了，看 redis-py 的源码，其等待的逻辑就是一个死循环，相关代码如下：

def acquire(self, blocking=None, blocking_timeout=None, token=None): # ...省略部分代码

    # 死循环等待获得锁
    while True: if self.do_acquire(token): self.local.token = token
            return True if not blocking: return False next_try_at = mod_time.time() + sleep
        if stop_trying_at is not None and next_try_at > stop_trying_at: return False # 阻塞睡眠一段时间
        mod_time.sleep(sleep) 

简单而言，这种方式就是在客户端轮询，未获得锁时，就等待一段时间再尝试去获得锁，直到成功获得锁或等待超时，这种方式实现简单，但当并发量比较大时，轮询的方式会耗费比较多资源，影响服务器性能。

更好的一种方式是使用 Redis 发布订阅功能，当线程 B 获取锁失败时，订阅锁释放的消息，当线程 A 执行完业务释放锁时，会发送锁释放信息，线程 B 获得信息后，再去获取锁，这样就不需要一直轮询了，而是直接休眠等待到锁释放消息则可。

Redis 集群主从切换

比较复杂的项目会使用多个 Redis 服务构建集群，Redis 集群采用主从方式部署，简单而言，通过算法选择出 Redis 集群中的主节点，所有写操作都会落到主节点上，主节点会将指令记录在 buffer 中，再通过异步的方式将 buffer 中的指令同步到其他从节点，从节点执行相同的指令，便会获得与主节点相同的数据结构。

当我们基于 Redis 集群来构建分布式锁时，可能会出现主从切换导致锁丢失的问题。

依旧以例子来说明，客户端 A 通过 Redis 集群成功加锁，这个操作首先会发生在主节点，但由于某些问题，当前 Redis 集群的主节点 down 了，此时根据相应的算法，Redis 集群会从从节点中选出新的主节点，这个过程对客户端 A 而言是透明的，但如果在主从切换时，客户端 A 在旧主节点加锁的指令还未同步它就 down 了，那么新的主节点就不会有客户端 A 加速的信息，此时，如果有新的客户端 B 要加锁，便可以轻松加上。

Redis 集群脑裂脑裂

这次确实挺抽象的，简单而言，Redis 集群中因为网络问题，某些从节点无法感知到主节点了，此时这些从节点会认为主节点 down 了，便会选出新的主节点，而客户端却可以连接上两个主节点，从而会出现两个客户端拥有同一把锁的情况。

结尾复杂分布式系统中锁的问题一直是个设计难题，学无止境呀。

原文地址:https://mp.weixin.qq.com/s/Z7RAImH2xW60xdiWnIYCyQ

本文链接：https://my.lmcjl.com/post/11024.html