Redis分布式锁注意事项

淩亂°似流年 2022-02-01 07:57 263阅读 0赞

## 基于 Redis 的分布式锁的安全性分析 ##

1.1 预防死锁  
我们看下面这个典型死锁场景。

> 一个客户端获取锁成功，但是在释放锁之前崩溃了，此时该客户端实际上已经失去了对公共资源的操作权，但却没有办法请求解锁（删除 Key-Value键值对），那么，它就会一直持有这个锁，而其它客户端永远无法获得锁。

我们的解决方案是：在加锁时为锁设置过期时间，当过期时间到达，Redis 会自动删除对应的 Key-Value，从而避免死锁。需要注意的是，这个过期时间需要结合具体业务综合评估设置，以保证锁的持有者能够在过期时间之内执行完相关操作并释放锁。

1.2 设置锁自动过期时间以预防死锁存在的隐患  
为了避免死锁，可利用 Redis 为锁数据（Key-Value）设置自动过期时间，虽然可以解决死锁的问题，但却存在隐患。

###### 我们看下面这个典型场景。 ######

1.  客户端 A 获取锁成功；
2.  客户端 A 在某个操作上阻塞了很长时间（对于 Java 而言，如发生 Full-GC）；
3.  过期时间到，锁自动释放；
4.  客户端 B 获取到了对应同一个资源的锁；
5.  客户端 A 从阻塞中恢复过来，认为自己依旧持有锁，继续操作同一个资源，导致互斥性失效。

这时我们可采取的解决方案见下。

1.  存在隐患的方案。第 5 步中，客户端 A 恢复后，可以比较下目前已经持有锁的时间，如果发现已经过期，则放弃对共享资源的操作即可避免互斥性失效的问题。但是，客户端 A 所在节点的时间和 Redis 节点的时间很可能不一致（比如客户端与 Redis 节点不在同一台服务器，而不同服务器时间通常不完全同步），因此，严格来讲，任何依赖两个节点时间比较结果的互斥性算法，都存在隐患。目前网上很多资料都采用了这种方案，鉴于其隐患，不推荐。
2.  可取的方案。既然比较时间不可取，那么，还可以比较 my\_random\_value，即客户端 A 恢复后，在操作共享资源前应比较目前自身所持有锁的 my\_random\_value 与 Redis 中存储的 my\_random\_value 是否一致，如果不相同，说明已经不再持有锁，则放弃对共享资源的操作以避免互斥性失效的问题。

1.3 解锁操作的原子性  
为了保证每次解锁操作都能正确进行，需要引入全局唯一变量 my\_random\_value。具体而言，解锁需要两步，先查询（GET）锁对应的 Value，与自己加锁时设置的 my\_random\_value 进行对比，如果相同，则可确认这把锁是自己加的，然后再发起解锁（DEL）。需要注意的是，GET 和 DEL 是两个操作，非原子性，那么解锁本身也会存在破坏互斥性的可能。

下面是典型场景。

1.  客户端 A 获取锁成功；
2.  客户端 A 访问共享资源；
3.  客户端 A 为了释放锁，先执行 GET 操作获取锁对应的随机字符串的值；
4.  客户端 A 判断随机字符串的值，与预期的值相等；
5.  客户端 A 由于某个原因阻塞了很长时间；
6.  过期时间到了，锁自动释放了；
7.  客户端 B 获取到了对应同一个资源的锁；
8.  客户端 A 从阻塞中恢复过来，执行 DEL 操纵，释放掉了客户端 B 持有的锁。

下面给出解决方案。  
如何保障解锁操作的原子性呢？在实践中，我总结出两种方案。

Jedis jedis = new Jedis("127.0.0.1", 6379);
    
    // “自旋”，等待锁
    String result = null;
    while (true)
    {
        // 申请锁，只有当“lock_name”不存在时才能申请成功，返回“OK",锁的过期时间设置为5s
        result = jedis.set("lock_name", "my_random_value", SET_IF_NOT_EXIST,
                SET_WITH_EXPIRE_TIME, 5000);
        if ("OK".equals(result))
        {
            break;
        }
    }
    
    // 监控锁对应的 Key，如果其它的客户端对这个 Key 进行了更改，那么本次事务会被取消。
    jedis.watch("lock_name");
    // 成功获取锁，则操作公共资源，自定义流程
    // to do something...
    
    // 释放锁之前，校验是否持有锁
    if (jedis.get("lock_name").equals("my_random_value"))
    {
        // 开启事务功能，
        Transaction multi = jedis.multi();
        // 模拟客户端阻塞10s，锁超时，自动清除
        try
        {
            Thread.sleep(5000);
        }
        catch (InterruptedException e)
        {
            e.printStackTrace();
        }
        // 客户端恢复，继续释放锁
        multi.del("lock_name");
        // 执行事务（如果其它的客户端对这个Key进行了更改，那么本次事务会被取消,不会执行)
        multi.exec();
    }
    
    // 释放资源
    jedis.unwatch();
    jedis.close();

1.  Redis 支持 Lua 脚本并保证其原子性，使用 Lua 脚本实现锁校验与释放，并使用 Redis 的 eval 函数执行 Lua 脚本，代码如下：

Jedis jedis = new Jedis("127.0.0.1", 6379);
    
    // “自旋”，等待锁
    String result = null;
    while (true)
    {
        // 申请锁，只有当“lock_name”不存在时才能申请成功，返回“OK"，锁的过期时间设置为 5s
        result = jedis.set("lock_name", "my_random_value", SET_IF_NOT_EXIST,
                SET_WITH_EXPIRE_TIME, 5000);
        if ("OK".equals(result))
        {
            break;
        }
    }
    // 成功获取锁，则操作公共资源，自定义流程
    // to do something...
    
    // Lua脚本，用于校验并释放锁     
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
    try
    {
        // 模拟客户端阻塞10s，锁超时，自动清除
        Thread.sleep(10000);
    }
    catch (InterruptedException e)
    {
        e.printStackTrace();
    }
    
    // 执行Lua脚本，校验并释放锁
    jedis.eval(script, Collections.singletonList("lock_name"),
            Collections.singletonList("my_random_value"));
    
    jedis.close();

1.4 Redis 节点故障后，主备切换的数据一致性  
考虑 Redis 节点宕机，如果长时间无法恢复，则导致锁服务长时间不可用。为了保证锁服务的可用性，通常的方案是给这个 Redis 节点挂一个 Slave（多个也可以），当 Master 节点不可用的时候，系统自动切到 Slave 上。但是由于 Redis 的主从复制（Replication）是异步的，这可能导致在宕机切换过程中丧失锁的安全性。  
我们看下典型场景。

1.  客户端 A 从 Master 获取了锁；
2.  Master 宕机了，存储锁的 Key 还没有来得及同步到 Slave 上；
3.  Slave 升级为 Master；
4.  客户端 B 从新的 Master 获取到了对应同一个资源的锁；
5.  客户端 A 和客户端 B 同时持有了同一个资源的锁，锁的安全性被打破。

###### 解决方案有两个。 ######

**方案1**，设想下，如果要避免上述情况，可以采用一个比较“土”的方法，即自认为持有锁的客户端在对敏感公共资源进行写操作前，先进行校验，确认自己是否确实持有锁，校验的方式前面已经介绍过——通过比较自己的 `my_random_value` 和 Redis 服务端中实际存储的 `my_random_value`。

显然，这里仍存在一个问题。如果校验完毕后，Master 数据尚未同步到 Slave 的情况下 Master 宕机，该如何是好？诚然，我们可以为 Redis 服务端设置较短的主从复置周期，以尽量避免上述情况出现，但是，隐患还是客观存在的。

**方案2**，针对该问题场景，Redis 的作者 Antirez 提出了 RedLock，其原理基于分布式一致性算法的核心理念：**多数派思想**。下面对 RedLock 做简要介绍。  
**1.5 RedLock 简要介绍**

1.  获取当前系统时间（毫秒数）。
2.  按顺序依次向 N 个 Redis 节点执行获取锁的操作。这个获取操作跟前面基于单 Redis 节点获取锁的过程相同，包含随机字符串 my\_random\_value，也包含过期时间（比如 PX 30000，即锁的有效时间）。为了保证在某个 Redis 节点不可用的时候算法能够继续运行，这个获取锁的操作还有一个超时时间（Time Out），它要远小于锁的有效时间（几十毫秒量级）。客户端在向某个 Redis 节点获取锁失败以后，应该立即尝试下一个 Redis 节点。这里的失败，应该包含任何类型的失败，比如该 Redis 节点不可用。
3.  计算获取锁的整个过程总共消耗了多长时间，计算方法是用当前时间减去第 1 步记录的时间。如果客户端从大多数 Redis 节点（>=N/2+1）成功获取到了锁，并且获取锁总共消耗的时间没有超过锁的有效时间（Lock Validity Time），那么这时客户端才认为最终获取锁成功；否则，认为最终获取锁失败。
4.  如果最终获取锁成功了，那么这个锁的有效时间应该重新计算，它等于最初的锁的有效时间减去第 3 步计算出来的获取锁消耗的时间。  
    如果最终获取锁失败了（可能由于获取到锁的 Redis 节点个数少于 N/2+1，或者整个获取锁的过程消耗的时间超过了锁的最初有效时间），那么客户端应该立即向所有 Redis 节点发起释放锁的操作（即前面介绍的 Redis Lua 脚本）。  
    \*\*我们再来了解下解锁步骤。\*\*上面描述的只是获取锁的过程，而释放锁的过程比较简单，即客户端向所有 Redis 节点发起释放锁的操作，不管这些节点在获取锁的时候成功与否。

**加锁核心方法为：**

jedis.set(String key, String value, String nxxx, String expx, int time)

**接下来说明下各个参数的意义。**

*  key：Redis 是 Key-Value 型数据库，key 具有唯一性，因此，用 key 作为锁。
 *  value：即例子中的 my\_random\_value，在 1.2 节《设置锁自动过期时间以预防死锁存在的隐患》中，我分析了隐患场景并给出了解决方案。为了保障可靠性，在解锁时，仅仅依赖 Key 是不够的，为了避免错误得释放锁，释放前需要进行校验，即根据 Key 取出 Value，将其与自己加锁时设置的 my\_random\_value 进行对比，以便确认是否是自己持有的锁。my\_random\_value 可以使用特定的随机算法生成，如 UUID.randomUUID().toString()。
 *  nxxx：根据 Redis 文档，这个参数填 NX，意思是 SET IF NOT EXIST，即当 Key 不存在时方可进行 SET 操作；若 Key 已经存在，则不做任何操作；
 *  expx：这个参数传的是 PX，表示给 Key 设置过期时间，具体时间由参数 time 决定。
 *  time：与参数 expx 相呼应，代表 Key 的过期时间，单位为毫秒。

#### 最后，我们做下小结。 ####

通过上述说明，set(…) 方法可以满足加锁的安全性，执行 set(…) 方法有两种结果。

1.  如果被争用的公共资源没有锁（即 Key 不存在），那么就进行加锁操作，并对锁设置个有效期，同时用具有特异性（一段时间内具有唯一性）Value 来标识加锁的客户端，以便解锁时进行校验。
2.  如果被争用的公共资源已经被加锁（即 Key 存在），则不做任何操作，通常的做法是等待锁释放，采用不断轮询的方式来确定锁是否释放，这种方式也被称为“自旋”等待。此外，还可以设置一个超时时间，如果在超时时间内未能加锁成功则退出。

## 解锁代码 ##

2.1 正确的解锁方式一  
Redis 支持 Lua 脚本并保证其原子性，使用 Lua 脚本实现锁校验与释放，并使用 Redis 的 eval() 函数执行 Lua 脚本，代码如下：

public class DistributedLock
    {
        // 释放锁成功标志
        private static final Long RELEASE_SUCCESS = 1L;
        /**
         * 释放锁
         * @param jedisPool jedis连接池
         * @param lockName 锁名，对应被争用的公共资源
         * @param myRandomValue 需保持全局唯一，以校验锁的持有者
         * @return 是否释放成功
         */
        public static boolean unLock(JedisPool jedisPool, String lockName, String myRandomValue)
        {
    
            Jedis jedis = null;
            // Lua脚本，用于校验并释放锁
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            try
            {
                jedis = jedisPool.getResource();
                Object result = jedis.eval(script, Collections.singletonList(lockName),
                        Collections.singletonList(myRandomValue));
    
                // 注意:如果脚本顺利执行将返回1，如果执行脚本中，其它的客户端对这个lockName对应的值进行了更改，那么将返回0
                if (RELEASE_SUCCESS.equals(result))
                {
                    return true;
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {
                if (null != jedis)
                {
                    jedis.close();
                }
            }
    
            return false;
        }
    }

从上面的示例代码可以看出，解锁操作只用了两行代码。

第一行使用了 Lua 脚本，其语义为通过 GET 命令访问参数 `KEYS[1]` 对应的锁，获得锁对应的 Value，并将其与参数 `ARGV[1]` 对比，如果相同则调用 DEL 命令删除 `KEYS[1]` 对应的键值对（即释放锁操作）。  
// Lua脚本，用于校验并释放锁

String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

第二行通过 Redis 的 eval() 函数执行 Lua 脚本，其中入参 lockName 赋值给参数 `KEYS[1]`，myRandomValue 赋值给 `ARGV[1]`，eval() 函数将 Lua 脚本交给 Redis 服务端执行。

jedis.eval(script, Collections.singletonList(lockName), Collections.singletonList(myRandomValue));

根据 Redis 官网文档说明，通过 eval() 执行 Lua 代码时，Lua 代码将被当成一个命令去执行（可保证原子性），并且直到 eval 命令执行完成，Redis 才会执行其他命令。因此，通过 Lua 脚本结合 eval 函数，可以科学得实现解锁操作的原子性，避免误解锁。

2.2 正确的解锁方式二  
使用 Redis 事务功能，通过 Watch 命令监控锁对应的 Key，释放锁则采用事务功能（Multi 命令），如果持有的锁已经因过期而释放（也可能释放后又被其它客户端持有），则 Key 对应的 Value 将改变，释放锁的事务将不会被执行，从而保证原子性，同时避免错误的释放锁，示例代码如下：

public class DistributedLock
    {
        private static final Long RELEASE_SUCCESS = 1L;
        /**
         * 释放锁
         * @param jedisPool jedis连接池
         * @param lockName 锁名，对应被争用的公共资源
         * @param myRandomValue 需保持全局唯一，以校验锁的持有者
         * @return 是否释放成功
         */
        public static boolean unLockII(JedisPool jedisPool, String lockName, String myRandomValue)
        {
            Jedis jedis = null;     
            try
            {
                jedis = jedisPool.getResource();
    
                // 监控锁对应的Key，如果其它的客户端对这个Key进行了更改，那么本次事务会被取消。
                jedis.watch(lockName);
                // 成功获取锁，则操作公共资源，自定义流程
                // to do something...
    
                // 校验是否持有锁
                if (myRandomValue.equals(jedis.get(lockName)))
                {
                    // 开启事务功能，
                    Transaction multi = jedis.multi();
                    // 释放锁
                    multi.del(lockName);
                    // 执行事务（如果其它的客户端对这个Key进行了更改，那么本次事务会被取消,不会执行)
                    // 如果正常执行，由于只有一个删除操作，返回的list将只有一个对象。
                    List<Object> result = multi.exec();
                    if (RELEASE_SUCCESS.equals(result.size()))
                    {
                        return true;
                    }
                }
            }
            catch (Exception e)
            {
                throw e;
            }
            finally
            {
                if (null != jedis)
                {
                    jedis.unwatch();
                    jedis.close();
                }
            }
    
            return false;
        }
    }