redis实现分布式锁的七种方式

什么是分布式锁

java中，我们在本地多线程访问共享资源时，为了数据的安全性，我们使用本地锁lock实现，但是如果访问分布式系统中的共享资源，不同主机之间同样需要一把锁，来保证分布式系统中资源的安全，称为分布式锁

分布式锁应该具有的特性

• 互斥性：同一时刻只能有一个线程获取锁
• 可重入性：一个线程获取锁之后，再次请求，可以拿到锁
• 安全性：锁只能被持有的线程释放，不能被其他线程释放
• 高性能和高可用：加锁和释放锁开销低，并且保持高可用性，避免宕机导致锁失效
• 锁的释放：保证锁的超时释放，避免死锁，防止不必要的资源浪费

方案一：setnx+expire

我们首先想到的就是setnx + expire命令，即先用setnx抢占到锁，抢到之后再给锁加一个过期时间，超时释放锁

setnx(set if not exists): key不存在，setnx设置成功返回1，key存在返回0

下面的伪代码：

key可以设置为对共享资源访问的key,value设置任意值

//加锁
if(if（jedis.setnx(lockKey,lockValue) == 1) {
    //设置过期时间
    expire(lockKey, 100);
    try {
        //处理业务
    } finally {
        //释放锁
        jedis.del(lockKey);
    }
}           

隐患：

这个方案中，setnx和expire两个命令分开了，不是原子操作。如果执行完setnx加锁，正要执行expire设置过期时间时，这时发生宕机了，那么这个锁就永远不会超时

方案二：setnx + value(value的值是系统时间+过期时间)

为了解决方案一中发生异常锁没有释放的场景，我们可以把过期时间作为setnx的value，如果别的线程加锁失败，获取value值比较一下时间。

伪代码如下：

getset（）：返回给定 key 的旧值。当 key 没有旧值时，即 key 不存在时，返回 nil 。当 key 存在但不是字符串类型时，返回一个错误。

//系统时间+设置的过期时间
long expires = System.currentTimeMillis() + expireTime; 
String expiresStr = String.valueOf(expires);
// 如果当前锁不存在，返回加锁成功
if (jedis.setnx(lockKey, expiresStr) == 1) {
        return true;
} 
// 如果锁已经存在，获取锁的过期时间
String currentValueStr = jedis.get(key_resource_id);
// 如果获取到的过期时间，小于系统当前时间，表示已经过期
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
     // 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间
    String oldValueStr = jedis.getSet(lockKey, expiresStr);
    if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
         // 考虑多线程并发的情况，只有一个线程的设置值和当前值相同，它才可以加锁
         return true;
    }
}        
    //其他情况，均返回加锁失败
    return false;
}           

这种方案在setnx里设置key的过期时间，解决了方案一发生异常锁得不到释放的问题

隐患

1. 过期时间是客户端自己生成的（System.currentTimeMillis()是当前系统的时间），必须要求分布式环境下，每个客户端的时间必须同步。
2. 如果锁过期的时候，多个线程并发进来，都执行jedis.getSet()，只能有一个线程加锁成功，线程之间的过期时间可能被覆盖
3. 锁可能被别的线程释放

方案三：使用Lua脚本(包含SETNX + EXPIRE两条指令)

使用lua脚本可以保证setnx和expire两条指令的原子性

伪代码:

lua脚本

if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then
   redis.call('expire',KEYS[1],ARGV[2])
else
   return 0
end;           

加锁代码

String lua= "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            " redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";   
Object result = jedis.eval(lua, Collections.singletonList(key_resource_id), Collections.singletonList(values));
//判断是否成功
return result.equals(1L);           

方案四：SET的扩展命令（SET EX PX NX）

上面方案是使用lua脚本保证两条命令的原子性，我们还可以使用redis的set指令(SET key value[EX seconds][PX milliseconds][NX|XX])这个也是原子的

SET key value[EX seconds][PX milliseconds][NX|XX]

NX :表示key不存在的时候，才能set成功，也即保证只有第一个客户端请求才能获得锁，而其他客户端请求只能等其释放锁，才能获取。

EX seconds :设定key的过期时间，时间单位是秒。

PX milliseconds: 设定key的过期时间，单位为毫秒

XX: 仅当key存在时设置值

伪代码：

if（jedis.set(key_resource_id, lock_value, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){
  }
  finally {
       jedis.del(key_resource_id); //释放锁
    }
}
           

隐患：

1. 锁过期释放了，业务还没执行完。假设线程a获取锁成功，一直在执行临界区的代码。但是100s过去后，它还没执行完。但是，这时候锁已经过期了，此时线程b又请求过来。显然线程b就可以获得锁成功，也开始执行临界区的代码。那么问题就来了，临界区的业务代码都不是严格串行执行的啦。
2. 锁被别的线程误删。假设线程a执行完后，去释放锁。但是它不知道当前的锁可能是线程b持有的（线程a去释放锁时，有可能过期时间已经到了，此时线程b进来占有了锁）。那线程a就把线程b的锁释放掉了，但是线程b业务代码可能都还没执行完

方案五：SET EX PX NX + 校验唯一随机值,再删除

上面说了可能被别的线程误删，那如果设置value的时候给每个线程设置一个随机值，在释放锁的时候，校验一下在释放

if（jedis.set(lockKey, uni_request_id, "NX", "EX", 100s) == 1）{ //加锁
    try {
        do something  //业务处理
    }catch(){
  }
  finally {
// 判断是不是当前线程加的锁和释放锁不是一个原子操作。
// 如果调用jedis.del()释放锁的时候，可能这把锁已经不属于当前客户端，会解除他人加的锁。
      if (uni_request_id.equals(jedis.get(lockKey))) {
        jedis.del(lockKey); //释放锁
      }
   }
}
           

方案六：开源框架redission

方案五可能存在的缺陷就是锁已经过期，但业务还没执行完成，我们可以想到的解决方案就是设置一个守护线程，每个一点时间检查锁是否还存在，存在就对延长锁的过期时间，不存在就释放锁

开源框架redission解决了这个问题

redission中增加了一个看门狗的机制，每隔一段时间去检查锁的状态，检查是否还持有锁，持有锁就延长锁的过期时间，这样就解决了锁过期释放，业务还没执行完的问题

方案七：分布式锁RedLock + redission

经过上面方案，还存在一个问题，如果线程一在Redis的master节点上拿到了锁，但是加锁的key还没同步到slave节点。恰好这时，master节点发生故障，一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦，但线程一也已经拿到锁了，锁就丧失了安全性

为了解决这个问题，redission的作者提出了RedLock,RedLock的核心思想

redis使用集群部署，这样存在多个master节点，不会同时宕机，需要确保在这多个master实例上，是与在Redis单实例，使用相同方法来获取和释放锁。

加锁步骤

• 按顺序向5个master节点请求加锁
• 根据设置的超时时间来判断，是不是要跳过该master节点。
• 如果大于等于3个节点加锁成功，并且使用的时间小于锁的有效期，即可认定加锁成功啦。
• 如果获取锁失败，解锁！