引言

分布式锁大家应该不生疏，在许多大厂面试的时候，面试官们都很喜爱问这个问题。

我们在系统中修正已有数据时，需要先读取，然后进行修正保存，此刻很简单遇到并发问题。由于修正和保存不是原子操作，在并发场景下，部分对数据的操作可能会丢掉。在单服务器系统我们常用本地锁来避免并发带来的问题，然而，当服务选用集群方法布置时，本地锁无法在多个服务器之间收效，这时候确保数据的共同性就需要分布式锁来完结。

Redis 之分布式锁的实现

完结

Redis 锁主要运用 Redis 的 setnx 指令。

  • 加锁指令：SETNX key value，当键不存在时，对键进行设置操作并回来成功，不然回来失利。KEY 是锁的唯一标识，一般按业务来决议命名。
  • 解锁指令：DEL key，经过删去键值对开释锁，以便其他线程能够经过 SETNX 指令来获取锁。
  • 锁超时：EXPIRE key timeout, 设置 key 的超时时刻，以确保即便锁没有被显式开释，锁也能够在必定时刻后主动开释，避免资源被永远锁住。

加锁/解锁伪代码如下：

if (setnx(key, 1) == 1){
    expire(key, 30)
    try {
        //TODO 业务逻辑
    } finally {
        del(key)
    }
}

上述锁完结方法存在一些问题：

1. SETNX 和 EXPIRE 非原子性

假如 SETNX 成功，在设置锁超时时刻后，服务器挂掉、重启或网络问题等，导致 EXPIRE 指令没有履行，锁没有设置超时时刻变成死锁。

Redis 之分布式锁的实现

能够运用 lua 脚本来解决这个问题，示例：

if (redis.call('setnx', KEYS[1], ARGV[1]) < 1)
then return 0;
end;
redis.call('expire', KEYS[1], tonumber(ARGV[2]));
return 1;
// 运用实例
EVAL "if (redis.call('setnx',KEYS[1],ARGV[1]) < 1) then return 0; end; redis.call('expire',KEYS[1],tonumber(ARGV[2])); return 1;" 1 key value 100

2. 锁误免除

假如线程 A 成功获取到了锁，而且设置了过期时刻 30 秒，但线程 A 履行时刻超过了 30 秒，锁过期主动开释，此刻线程 B 获取到了锁；随后 A 履行完结，线程 A 运用 DEL 指令来开释锁，但此刻线程 B 加的锁还没有履行完结，线程 A 实际开释的线程 B 加的锁。

Redis 之分布式锁的实现

经过在 value 中设置当时线程加锁的标识，在删去之前验证 key 对应的 value 判断锁是否是当时线程持有。可生成一个 UUID 标识当时线程，运用 lua 脚本做验证标识和解锁操作。

// 加锁
String uuid = UUID.randomUUID().toString().replaceAll("-","");
SET key uuid NX EX 30
// 解锁
if (redis.call('get', KEYS[1]) == ARGV[1])
    then return redis.call('del', KEYS[1])
else return 0
end

3. 超时解锁导致并发

假如线程 A 成功获取锁并设置过期时刻 30 秒，但线程 A 履行时刻超过了 30 秒，锁过期主动开释，此刻线程 B 获取到了锁，线程 A 和线程 B 并发履行。

Redis 之分布式锁的实现

A、B 两个线程发生并发显然是不被答应的，一般有两种方法解决该问题：

  • 将过期时刻设置足够长，确保代码逻辑在锁开释之前能够履行完结。
  • 为获取锁的线程添加守护线程，为即将过期但未开释的锁添加有效时刻。（主动续期）

Redis 之分布式锁的实现

4. 不行重入

当线程在持有锁的情况下再次恳求加锁，假如一个锁支持一个线程多次加锁，那么这个锁便是可重入的。假如一个不行重入锁被再次加锁，由于该锁已经被持有，再次加锁会失利。Redis 可经过对锁进行重入计数，加锁时加 1，解锁时减 1，当计数归 0 时开释锁。

在本地记载记载重入次数，如 Java 中运用 ThreadLocal 进行重入次数统计，简单示例代码：

private static ThreadLocal<Map<String, Integer>> LOCKERS = ThreadLocal.withInitial(HashMap::new);
// 加锁
public boolean lock(String key) {
  Map<String, Integer> lockers = LOCKERS.get();
  if (lockers.containsKey(key)) {
    lockers.put(key, lockers.get(key) + 1);
    return true;
  } else {
    if (SET key uuid NX EX 30) {
      lockers.put(key, 1);
      return true;
    }
  }
  return false;
}
// 解锁
public void unlock(String key) {
  Map<String, Integer> lockers = LOCKERS.get();
  if (lockers.getOrDefault(key, 0) <= 1) {
    lockers.remove(key);
    DEL key
  } else {
    lockers.put(key, lockers.get(key) - 1);
  }
}

本地记载重入次数虽然高效，但假如考虑到过期时刻和本地、Redis 共同性的问题，就会添加代码的复杂性。另一种方法是 Redis Map 数据结构来完结分布式锁，既存锁的标识也对重入次数进行计数。Redission 加锁示例：

// 假如 lock_key 不存在
if (redis.call('exists', KEYS[1]) == 0)
then
    // 设置 lock_key 线程标识 1 进行加锁
    redis.call('hset', KEYS[1], ARGV[2], 1);
    // 设置过期时刻
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
    end;
// 假如 lock_key 存在且线程标识是当时欲加锁的线程标识
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1)
    // 自增
    then redis.call('hincrby', KEYS[1], ARGV[2], 1);
    // 重置过期时刻
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return nil;
    end;
// 假如加锁失利，回来锁剩余时刻
return redis.call('pttl', KEYS[1]);

5. 无法等候锁开释

上述指令履行都是立即回来的，假如客户端能够等候锁开释就无法运用。

  • 能够经过客户端轮询的方法解决该问题，当未获取到锁时，等候一段时刻从头获取锁，直到成功获取锁或等候超时。这种方法比较耗费服务器资源，当并发量比较大时，会影响服务器的功率。
  • 另一种方法是运用 Redis 的发布订阅功用，当获取锁失利时，订阅锁开释消息，获取锁成功后开释时，发送锁开释消息。如下：

Redis 之分布式锁的实现

Redis 还有 Redlock 这种分布式锁

由于这种方法的运用有争议（极点情况存在问题），本文暂不介绍！

集群

1. 主备切换

为了确保 Redis 的可用性，一般选用主从方法布置。主从数据同步有异步和同步两种方法，Redis 将指令记载在本地内存 buffer 中，然后异步将 buffer 中的指令同步到从节点，从节点一边履行同步的指令流来到达和主节点共同的状况，一边向主节点反应同步情况。

在包含主从形式的集群布置方法中，当主节点挂掉时，从节点会取而代之，但客户端无显着感知。当客户端 A 成功加锁，指令还未同步，此刻主节点挂掉，从节点提高为主节点，新的主节点没有锁的数据，当客户端 B 加锁时就会成功。

Redis 之分布式锁的实现

2. 集群脑裂

集群脑裂指由于网络问题，导致 Redis master 节点跟 slave 节点和 sentinel 集群处于不同的网络分区，由于 sentinel 集群无法感知到 master 的存在，所以将 slave 节点提高为 master 节点，此刻存在两个不同的 master 节点。Redis Cluster 集群布置方法同理。

当不同的客户端连接不同的 master 节点时，两个客户端能够同时具有同一把锁。如下：

Redis 之分布式锁的实现

结语

Redis 以高性能著称，但运用其完结分布式锁来解决并发仍存在一些困难。Redis 分布式锁只能作为一种缓解并发的手法，假如要完全解决并发问题，仍需要数据库的防并发手法。