redis分布式锁原理与实现

╰+哭是因爲堅強的太久メ 2022-11-20 01:57 237阅读 0赞

## 分布式锁原理 ##

分布式锁，是控制分布式系统之间同步访问共享资源的一种方式。在分布式系统中，常常需要协调他们的动作。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，在这种情况下，便需要使用到分布式锁。

使用setnx、getset、expire、del这4个redis命令实现：

1.  setnx 是『SET if Not eXists』(如果不存在，则 SET)的简写。 **命令格式**：SETNX key value；使用：只在键 key 不存在的情况下，将键 key 的值设置为 value 。若键 key 已经存在， 则 SETNX 命令不做任何动作。**返回值**：命令在设置成功时返回 1 ，设置失败时返回 0 。
2.  getset **命令格式**：GETSET key value，将键 key 的值设为 value ，并返回键 key 在被设置之前的旧的value。**返回值**：如果键 key 没有旧值， 也即是说， 键 key 在被设置之前并不存在， 那么命令返回 nil 。当键 key 存在但不是字符串类型时，命令返回一个错误。
3.  expire **命令格式**：EXPIRE key seconds，**使用**：为给定 key 设置生存时间，当 key 过期时(生存时间为 0 )，它会被自动删除。**返回值**：设置成功返回 1 。 当 key 不存在或者不能为 key 设置生存时间时(比如在低于 2.1.3 版本的 Redis 中你尝试更新 key 的生存时间)，返回 0 。
4.  del **命令格式**：DEL key \[key …\]，**使用**：删除给定的一个或多个 key ，不存在的 key 会被忽略。**返回值**：被删除 key 的数量。

### redis 分布式锁原理一 ###

原理图如下

![å¨è¿éæå¥å¾çæè¿°][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhem91MQ_size_16_color_FFFFFF_t_70]

过程分析：

1.  A尝试去获取锁lockkey，通过setnx(lockkey,currenttime+timeout)命令，对lockkey进行setnx,将value值设置为当前时间+锁超时时间；
2.  如果返回值为1，说明redis服务器中还没有lockkey，也就是没有其他用户拥有这个锁，A就能获取锁成功；
3.  在进行相关业务执行之前，先执行expire(lockkey)，对lockkey设置有效期，防止死锁。因为如果不设置有效期的话，lockkey将一直存在于redis中，其他用户尝试获取锁时，执行到setnx(lockkey,currenttime+timeout)时，将不能成功获取到该锁；
4.  执行相关业务；
5.  释放锁，A完成相关业务之后，要释放拥有的锁，也就是删除redis中该锁的内容，del(lockkey)，接下来的用户才能进行重新设置锁新值。

代码实现：

public void redis1() {
            log.info("关闭订单定时任务启动");
            long lockTimeout = Long.parseLong(PropertiesUtil.getProperty("lock.timeout", "5000"));
            //这个方法的缺陷在这里，如果setnx成功后，锁已经存到Redis里面了，服务器异常关闭重启，将不会执行closeOrder，也就不会设置锁的有效期，这样的话锁就不会释放了，就会产生死锁
            Long setnxResult = RedisShardedPoolUtil.setnx(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, String.valueOf(System.currentTimeMillis() + lockTimeout));
            if (setnxResult != null && setnxResult.intValue() == 1) {
                //如果返回值为1，代表设置成功，获取锁
                closeOrder(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
            } else {
                log.info("没有获得分布式锁:{}", Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
            }
            log.info("关闭订单定时任务结束");
        }
    private void closeOrder(String lockName) {
            //对锁设置有效期
            RedisShardedPoolUtil.expire(lockName, 5);//有效期为5秒，防止死锁
            log.info("获取锁:{},ThreadName:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName());
            //执行业务
            int hour = Integer.parseInt(PropertiesUtil.getProperty("close.order.task.time.hour", "2"));
            iOrderService.closeOrder(hour);
            //执行完业务后，释放锁
            RedisShardedPoolUtil.del(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
            log.info("释放锁:{},ThreadName:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName());
            log.info("=================================");
        }

缺陷：  
如果A在setnx成功后，A成功获取锁了，也就是锁已经存到Redis里面了，此时服务器异常关闭或是重启，将不会执行closeOrder，也就不会设置锁的有效期，这样的话锁就不会释放了，就会产生死锁。

解决方法：  
关闭Tomcat有两种方式，一种通过温柔的执行shutdown关闭，一种通过kill杀死进程关闭

//通过温柔的执行shutdown关闭时，以下的方法会在关闭前执行，即可以释放锁，而对于通过kill杀死进程关闭时，以下方法不会执行，即不会释放锁
       //这种方式释放锁的缺点在于，如果关闭的锁过多，将造成关闭服务器耗时过长
        @PreDestroy
        public void delLock() {
            RedisShardedPoolUtil.del(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
        }

### redis 分布式锁原理2(优化版) ###

为了解决原理1中会出现的死锁问题，提出原理2双重防死锁，可以更好解决死锁问题。  
原理图如下：

![å¨è¿éæå¥å¾çæè¿°][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhem91MQ_size_16_color_FFFFFF_t_70 1]

过程分析：

1.  当A通过setnx(lockkey,currenttime+timeout)命令能成功设置lockkey时，即返回值为1，过程与原理1一致；
2.  当A通过setnx(lockkey,currenttime+timeout)命令不能成功设置lockkey时，这是不能直接断定获取锁失败；因为我们在设置锁时，设置了锁的超时时间timeout，当当前时间大于redis中存储键值为lockkey的value值时，可以认为上一任的拥有者对锁的使用权已经失效了，A就可以强行拥有该锁；具体判定过程如下；
3.  A通过get(lockkey)，获取redis中的存储键值为lockkey的value值，即获取锁的相对时间lockvalueA
4.  lockvalueA!=null && currenttime>lockvalue，A通过当前的时间与锁设置的时间做比较，如果当前时间已经大于锁设置的时间临界，即可以进一步判断是否可以获取锁，否则说明该锁还在被占用，A就还不能获取该锁，结束，获取锁失败；
5.  步骤4返回结果为true后，通过getSet设置新的超时时间，并返回旧值lockvalueB，以作判断，因为在分布式环境，在进入这里时可能另外的进程获取到锁并对值进行了修改，只有旧值与返回的值一致才能说明中间未被其他进程获取到这个锁
6.  lockvalueB == null || lockvalueA==lockvalueB，判断：若果lockvalueB为null，说明该锁已经被释放了，此时该进程可以获取锁；旧值与返回的lockvalueB一致说明中间未被其他进程获取该锁，可以获取锁；否则不能获取锁，结束，获取锁失败。

代码实现：

public void redis2() {
            log.info("关闭订单定时任务启动");
            long lockTimeout = Long.parseLong(PropertiesUtil.getProperty("lock.timeout", "5000"));
            Long setnxResult = RedisShardedPoolUtil.setnx(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, String.valueOf(System.currentTimeMillis() + lockTimeout));
            if (setnxResult != null && setnxResult.intValue() == 1) {
                //如果返回值为1，代表设置成功，获取锁
                closeOrder(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
            } else {
                //未获取到锁，继续判断，判断时间戳，看是否可以重置并获取到锁
                String lockValueStr = RedisShardedPoolUtil.get(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
                //通过当前的时间与锁设置的时间做比较，如果当前时间已经大于锁设置的时间临界，即可以进一步判断是否可以获取锁，否则说明该锁还在被占用，不能获取该锁
                if (lockValueStr != null && System.currentTimeMillis() > Long.parseLong(lockValueStr)) {
                    //通过getSet设置新的超时时间，并返回旧值，以作判断，因为在分布式环境，在进入这里时可能另外的进程获取到锁并对值进行了修改，只有旧值与返回的值一致才能说明中间未被其他进程获取到这个锁
                    String getSetResult = RedisShardedPoolUtil.getSet(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, String.valueOf(System.currentTimeMillis() + lockTimeout));
                    //再次用当前时间戳getset。
                    //返回给定的key的旧值，与旧值判断，是否可以获取锁
                    //当key没有旧值时，即key不存在时，返回nil ->获取锁
                    //这里我们set了一个新的value值，获取旧的值。
                    //若果getSetResult为null，说明该锁已经被释放了，此时该进程可以获取锁；旧值与返回的getSetResult一致说明中间未被其他进程获取该锁，可以获取锁
                    if (getSetResult == null || (getSetResult != null && StringUtils.equals(lockValueStr, getSetResult))) {
                        //真正获取到锁
                        closeOrder(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
                    } else {
                        log.info("没有获得分布式锁:{}", Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
                    }
                } else {
                    log.info("没有获得分布式锁:{}", Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
                }
            }
            log.info("关闭订单定时任务结束");
        }
        private void closeOrder(String lockName) {
            //对锁设置有效期
            RedisShardedPoolUtil.expire(lockName, 5);//有效期为5秒，防止死锁
            log.info("获取锁:{},ThreadName:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName());
            //执行业务
            int hour = Integer.parseInt(PropertiesUtil.getProperty("close.order.task.time.hour", "2"));
            iOrderService.closeOrder(hour);
            //执行完业务后，释放锁
            RedisShardedPoolUtil.del(Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK);
            log.info("释放锁:{},ThreadName:{}",Const.REDIS_LOCK.CLOSE_ORDER_TASK_LOCK, Thread.currentThread().getName());
            log.info("=================================");
        }

优化点：

*  加入了超时时间判断锁是否超时了，及时A在成功设置了锁之后，服务器就立即出现宕机或是重启，也不会出现死锁问题；因为B在尝试获取锁的时候，如果不能setnx成功，会去获取redis中锁的超时时间与当前的系统时间做比较，如果当前的系统时间已经大于锁超时时间，说明A已经对锁的使用权失效，B能继续判断能否获取锁，解决了redis分布式锁的死锁问题。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhem91MQ_size_16_color_FFFFFF_t_70]: /images/20221120/c75f10e8a9db4548b3681e2f63a97d18.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhem91MQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20190303142906677.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rhem91MQ==,size_16,color_FFFFFF,t_70