分布式|使用Redis SETNX 命令实现分布式锁

ゞ浴缸里的玫瑰 2023-06-12 11:07 41阅读 0赞

**使用分布式锁要满足几个条件**

1.  系统是一个分布式系统（关键是分布式，单机的可以使用ReentrantLock或者synchronized代码块来实现）
2.  共享资源（各个系统访问同一个资源，资源的载体可能是传统关系型数据库或者NoSQL）
3.  同步访问（即有很多个进程同事访问同一个共享资源。没有同步访问，谁管你资源竞争不竞争）

**分布式锁实现方式****  **

**1、数据库的唯一索引**

获得锁时向表中插入一条记录，释放锁时删除这条记录。

**2、Redis的SETNX指令**

与数据库唯一索引类似，缓存性能更好。缺点如果redis发生单点故障，则分布式锁无法使用。

**3、Redis的RedLock算法**

使用多个Redis实例来实现分布式锁，基于`set nx ex`实现

**4、基于Redis的Redisson分布式可重入锁**

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java的Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。基于setnx+redis事务机制+redis的pub/sub机制实现

**5、Zookeeper的有序节点**

1）获取锁时，在/lock锁目录下创建临时且有序的子节点；

2）客户端判断创建的子节点为当前子节点列表中序号最小的子节点，则获得锁，否则监听前一个子节点，获得子节点的变更通知后重复此步骤直至获得锁。

3）执行业务代码，完成后，删除对应的子节点。

4）会话超时自动释放临时节点

**下面介绍使用Redis的SETNX 实现分布式锁方法。**

**命令格式**

>  SETNX key value

将 key 的值设为 value，当且仅当 key 不存在。若给定的 key 已经存在，则 SETNX 不做任何动作。SETNX 是SET if Not eXists的简写。

## 返回值 ##

返回整数，具体为- 1，当 key 的值被设置- 0，当 key 的值没被设置

## 例子 ##

> redis> SETNX mykey “hello” 
> 
>  (integer) 1

多个进程执行以下Redis命令：

SETNX lock.foo <current Unix time + lock timeout + 1>

如果 SETNX 返回1，说明该进程获得锁，SETNX将键 lock.foo 的值设置为锁的超时时间（当前时间 + 锁的有效时间）。    如果 SETNX 返回0，说明其他进程已经获得了锁，进程不能进入临界区。进程可以在一个循环中不断地尝试 SETNX 操作，以获得锁。

##     ##

## **解决死锁** ##

考虑一种情况，如果进程获得锁后，断开了与 Redis 的连接（可能是进程挂掉，或者网络中断），如果没有有效的释放锁的机制，那么其他进程都会处于一直等待的状态，即出现“死锁”。

上面在使用 SETNX 获得锁时，我们将键 lock.foo 的值设置为锁的有效时间，进程获得锁后，其他进程还会不断的检测锁是否已超时，如果超时，那么等待的进程也将有机会获得锁。

然而，锁超时时，我们不能简单地使用 DEL 命令删除键 lock.foo 以释放锁。考虑以下情况，进程P1已经首先获得了锁 lock.foo，然后进程P1挂掉了。进程P2，P3正在不断地检测锁是否已释放或者已超时，执行流程如下：

*  P2和P3进程读取键 lock.foo 的值，检测锁是否已超时（通过比较当前时间和键 lock.foo 的值来判断是否超时）
 *  P2和P3进程发现锁 lock.foo 已超时
 *  P2执行 DEL lock.foo命令
 *  P2执行 SETNX lock.foo命令，并返回1，即P2获得锁
 *  P3执行 DEL lock.foo命令将P2刚刚设置的键 lock.foo 删除（这步是由于P3刚才已检测到锁已超时）
 *  P3执行 SETNX lock.foo命令，并返回1，即P3获得锁
 *  P2和P3同时获得了锁

从上面的情况可以得知，在检测到锁超时后，进程不能直接简单地执行 DEL 删除键的操作以获得锁。

为了解决上述算法可能出现的多个进程同时获得锁的问题，我们再来看以下的算法。    我们同样假设进程P1已经首先获得了锁 lock.foo，然后进程P1挂掉了。接下来的情况：

*  进程P4执行 SETNX lock.foo 以尝试获取锁
 *  由于进程P1已获得了锁，所以P4执行 SETNX lock.foo 返回0，即获取锁失败
 *  P4执行 GET lock.foo 来检测锁是否已超时，如果没超时，则等待一段时间，再次检测
 *  如果P4检测到锁已超时，即当前的时间大于键 lock.foo 的值，P4会执行以下操作
 *      GETSET lock.foo <current Unix timestamp + lock timeout + 1>
 *  
 *  由于 GETSET 操作在设置键的值的同时，还会返回键的旧值，通过比较键 lock.foo 的旧值是否小于当前时间，可以判断进程是否已获得锁
 *  假如另一个进程P5也检测到锁已超时，并在P4之前执行了 GETSET 操作，那么P4的 GETSET 操作返回的是一个大于当前时间的时间戳，这样P4就不会获得锁而继续等待。注意到，即使P4接下来将键 lock.foo 的值设置了比P5设置的更大的值也没影响。

另外，值得注意的是，在进程释放锁，即执行 DEL lock.foo 操作前，需要先判断锁是否已超时。如果锁已超时，那么锁可能已由其他进程获得，这时直接执行 DEL lock.foo 操作会导致把其他进程已获得的锁释放掉。

**程序代码**

public class RedisLock {
        private RedisTemplate redisTemplate;
        private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
        private String lockKey;
        /**
         * 锁超时时间，防止线程在入锁以后，无限的执行等待
         */
        private int expireMsecs = 60 * 1000;
        /**
         * 锁等待时间，防止线程饥饿
         */
        private int timeoutMsecs = 10 * 1000;
        private volatile boolean locked = false;
    
        /**
         * Detailed constructor.
         *
         */
        public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
            this.redisTemplate = redisTemplate;
            this.lockKey = lockKey + "_lock";
            this.timeoutMsecs = timeoutMsecs;
            this.expireMsecs = expireMsecs;
        }
    
        /**
         * @return lock key
         */
        public String getLockKey() {
            return lockKey;
        }
    
        /**
         * 获得 lock.
         * 实现思路: 主要是使用了redis 的setnx命令,缓存了锁.
         * reids缓存的key是锁的key,所有的共享, value是锁的到期时间(注意:这里把过期时间放在value了,没有时间上设置其超时时间)
         * 执行过程:
         * 1.通过setnx尝试设置某个key的值,成功(当前没有这个锁)则返回,成功获得锁
         * 2.锁已经存在则获取锁的到期时间,和当前时间比较,超时的话,则设置新的值
         *
         * @return true if lock is acquired, false acquire timeouted
         * @throws InterruptedException in case of thread interruption
         */
        public synchronized boolean lock() throws InterruptedException {
            int timeout = timeoutMsecs;
            while (timeout >= 0) {
                long expires = System.currentTimeMillis() + expireMsecs + 1;
                String expiresStr = String.valueOf(expires); //锁到期时间
                if (this.setNX(lockKey, expiresStr)) {
                    // lock acquired
                    locked = true;
                    return true;
                }
    
                String currentValueStr = this.get(lockKey); //redis里的时间
                if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
                    //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
                    // lock is expired
    
                    String oldValueStr = this.getSet(lockKey, expiresStr);
                    //获取上一个锁到期时间，并设置现在的锁到期时间，
                    //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
                    if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                        //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受
    
                        //[分布式的情况下]:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
                        // lock acquired
                        locked = true;
                        return true;
                    }
                }
                timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
    
                /*
                    延迟100 毫秒,  这里使用随机时间可能会好一点,可以防止饥饿进程的出现,即,当同时到达多个进程,
                    只会有一个进程获得锁,其他的都用同样的频率进行尝试,后面有来了一些进行,也以同样的频率申请锁,这将可能导致前面来的锁得不到满足.
                    使用随机的等待时间可以一定程度上保证公平性
                 */
                Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
    
            }
            return false;
        }
    
    
        /**
         * Acqurired lock release.
         */
        public synchronized void unlock() {
            if (locked) {
                redisTemplate.delete(lockKey);
                locked = false;
            }
        }
    
    }

1、为什么不直接使用expire设置超时时间，而将时间的毫秒数其作为value放在redis中？

如下面的方式，把超时的交给redis处理：

因为是分布式的环境下，可以在前一个锁失效的时候，有两个进程进入到锁超时的判断。如：

C0超时了，还持有锁,C1/C2同时请求进入了方法里面

C1/C2获取到了C0的超时时间

C1使用getSet方法

C2也执行了getSet方法

假如我们不加 oldValueStr.equals(currentValueStr) 的判断，将会C1/C2都将获得锁，加了之后，能保证C1和C2只能一个能获得锁，一个只能继续等待。

**注意：****这里可能导致超时时间不是其原本的超时时间，C1的超时时间可能被C2覆盖了，但是他们相差的毫秒及其小，这里忽略了**

参考资料：

https://github.com/redisson/redisson/wiki/8.-分布式锁和同步器

https://blog.csdn.net/liubenlong007/article/details/53782934