使用redis+lua脚本实现分布式锁

矫情吗；* 2022-05-27 08:19 473阅读 0赞

### 分布式锁的应用场景 ###

当多个机器（多个进程）会对同一条数据进行修改时，并且要求这个修改是原子性的。这里有两个限定：

>  *  多个进程之间的竞争，意味着JDK自带的锁失效；
>  *  原子性修改，意味着数据是有状态的，修改前后有依赖。

分布式锁的实现条件：

>  *  高性能(加、解锁时高性能)
>  *  可以使用阻塞锁与非阻塞锁。
>  *  不能出现死锁。
>  *  可用性(不能出现节点 down 掉后加锁失败)。

### Redis使用分布式锁 ###

本文将先介绍Redis的实现方式，当然Redis实现分布式锁有多种方案，本文介绍一种基于lua脚本的实现方案。后面笔者会介绍分布式锁的其他实现(基于 DB 的唯一索引和基于 ZK 的临时有序节点)。

本方案中Redis的实现主要基于setnx 和给予一个超时时间（防止释放锁失败）。  
多个尝试获取锁的客户端使用同一个key做为目标数据的唯一键，value为锁的期望超时时间点；  
首先进行一次setnx命令，尝试获取锁，如果获取成功，则设置锁的最终超时时间（以防在当前进程获取锁后奔溃导致锁无法释放）

这里利用 Redis set key 时的一个 NX 参数可以保证在这个 key 不存在的情况下写入成功。并且再加上 EX 参数可以让该 key 在超时之后自动删除。

注意：此处使用Jedis的如下方法，该命令可以保证 NX EX 的原子性。

**一定不要把两个命令(NX EX)分开执行，如果在 NX 之后程序出现问题就有可能产生死锁。**

String set(String key, String value, String nxxx, String expx, long time);

### 非阻塞锁 ###

public  boolean tryLock(String key, String request) {
        String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);
        if (LOCK_MSG.equals(result)){
            return true ;
        }else {
            return false ;
        }
    }

### 阻塞锁 ###

同时也可以实现一个阻塞锁：

//一直阻塞
    public void lock(String key, String request) throws InterruptedException {
        for (;;){
            String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);
            if (LOCK_MSG.equals(result)){
                break ;
            }
     //防止一直消耗 CPU 	
            Thread.sleep(DEFAULT_SLEEP_TIME) ;
        }
    }
     //自定义阻塞时间
     public boolean lock(String key, String request,int blockTime) throws InterruptedException {
        while (blockTime >= 0){
            String result = this.jedis.set(LOCK_PREFIX + key, request, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, 10 * TIME);
            if (LOCK_MSG.equals(result)){
                return true ;
            }
            blockTime -= DEFAULT_SLEEP_TIME ;
            Thread.sleep(DEFAULT_SLEEP_TIME) ;
        }
        return false ;
    }

### 解锁 ###

解锁也很简单，其实就是把这个 key 删掉就万事大吉了，比如使用 del key 命令。

但现实往往没有那么 easy。

如果进程 A 获取了锁设置了超时时间，但是由于执行周期较长导致到了超时时间之后锁就自动释放了。这时进程 B 获取了该锁执行很快就释放锁。这样就会出现进程 B 将进程 A 的锁释放了。

所以最好的方式是在每次解锁时都需要判断锁是否是自己的。

这时就需要结合加锁机制一起实现了。

加锁时需要传递一个参数，将该参数作为这个 key 的 value，这样每次解锁时判断 value 是否相等即可。

**为了更好的健壮性，将该操作封装为一个lua脚本，这样即可保证其原子性**

public  boolean unlock(String key,String request){
        //lua script
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = null ;
        if (jedis instanceof Jedis){
            result = ((Jedis)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request));
        }else if (jedis instanceof JedisCluster){
            result = ((JedisCluster)this.jedis).eval(script, Collections.singletonList(LOCK_PREFIX + key), Collections.singletonList(request));
        }else {
            //throw new RuntimeException("instance is error") ;
            return false ;
        }
        if (UNLOCK_MSG.equals(result)){
            return true ;
        }else {
            return false ;
        }
    }

### 该方案的缺点 ###

#### key 超时之后业务并没有执行完毕但却自动释放锁了 ####

#### 如运行在Redis的集群，当进程1对master节点写入了锁，此时master节点宕机。slave节点提升为master而刚刚写入master的锁还未同步，此时进程2也将能够获取锁成功。 ####