集群环境下Redis分布式锁的正确姿势

「爱情、让人受尽委屈。」 2022-10-01 11:51 178阅读 0赞

# RedLock算法的实现思路 #

antirez提出的redlock算法实现思路大概是这样的。

**客户端按照下面的步骤来获取锁：**

1.  获取当前时间的毫秒数T1。
2.  按顺序依次向N个Redis节点执行获取锁的操作。这个获取锁的操作和上一篇中基于单Redis节点获取锁的过程相同。包括唯一UUID作为Value以及锁的过期时间(expireTime)。为了保证在某个在某个Redis节点不可用的时候算法能够继续运行，这个获取锁的操作还需要一个超时时间。它应该远小于锁的过期时间。客户端向某个Redis节点获取锁失败后，应立即尝试下一个Redis节点。这里失败包括Redis节点不可用或者该Redis节点上的锁已经被其他客户端持有。
3.  计算整个获取锁过程的总耗时。即当前时间减去第一步记录的时间。计算公司为T2=now()- T1。如果客户端从大多数Redis节点(>N/2 +1)成功获取到锁。并且获取锁总共消耗的时间小于锁的过期时间（即T2<expireTime）。则认为客户端获取锁成功，否则，认为获取锁失败
4.  如果获取锁成功，需要重新计算锁的过期时间。它等于最初锁的有效时间减去第三步计算出来获取锁消耗的时间，即expireTime - T2
5.  如果最终获取锁失败，那么客户端立即向所有Redis系欸但发起释放锁的操作。（和上一篇释放锁的逻辑一样）

**虽然说RedLock算法可以解决单点Redis分布式锁的安全性问题，但如果集群中有节点发生崩溃重启，还是会锁的安全性有影响的。具体出现问题的场景如下：**

假设一共有5个Redis节点：A, B, C, D, E。设想发生了如下的事件序列：

1.  客户端1成功锁住了A, B, C，获取锁成功（但D和E没有锁住）
2.  节点C崩溃重启了，但客户端1在C上加的锁没有持久化下来，丢失了
3.  节点C重启后，客户端2锁住了C, D, E，获取锁成功

这样，客户端1和客户端2同时获得了锁（针对同一资源）。针对这样场景，解决方式也很简单，也就是让Redis崩溃后延迟重启，并且这个延迟时间大于锁的过期时间就好。这样等节点重启后，所有节点上的锁都已经失效了。也不存在以上出现2个客户端获取同一个资源的情况了。

RedLock安全性和稳定性好，但要说完全没有问题不是。例如，如果客户端获取锁成功后，如果访问共享资源操作执行时间过长，导致锁过期了，后续客户端获取锁成功了，这样在同一个时刻又出现了2个客户端获得了锁的情况。所以针对分布式锁的应用的时候需要多测试。服务器台数越多，出现不可预期的情况也越多。如果客户端获取锁之后，在上面第三步发生了GC得情况导致GC完成后，锁失效了，这样同时也使得同一时间有2个客户端获得了锁。如果系统对共享资源有非常严格要求得情况下，还是建议需要做数据库锁得得方案来补充。如飞机票或火车票座位得情况。对于一些抢购获取，针对偶尔出现超卖，后续可以人为沟通置换得方式采用分布式锁得方式没什么问题。因为可以绝大部分保证分布式锁的安全性。

# 分布式场景下基于Redis实现分布式锁的正确姿势 #

目前redisson包已经有对redlock算法封装，接下来就具体看看使用redisson包来实现分布式锁的正确姿势。

**具体实现代码如下代码所示：**

public interface DistributedLock {
        /**
         * 获取锁
         * @return 锁标识
         */
        String acquire();
    
        /**
         * 释放锁
         * @param indentifier
         * @return
         */
        boolean release(String indentifier);
    }
    
    public class RedisDistributedRedLock implements DistributedLock {
    
        /**
         * redis 客户端
         */
        private RedissonClient redissonClient;
    
        /**
         * 分布式锁的键值
         */
        private String lockKey;
    
        private RLock redLock;
    
        /**
         * 锁的有效时间 10s
         */
        int expireTime = 10 * 1000;
    
        /**
         * 获取锁的超时时间
         */
        int acquireTimeout  = 500;
    
        public RedisDistributedRedLock(RedissonClient redissonClient, String lockKey) {
            this.redissonClient = redissonClient;
            this.lockKey = lockKey;
        }
    
        @Override
        public String acquire() {
            redLock = redissonClient.getLock(lockKey);
            boolean isLock;
            try{
                isLock = redLock.tryLock(acquireTimeout, expireTime, TimeUnit.MILLISECONDS);
                if(isLock){
                    System.out.println(Thread.currentThread().getName() + " " + lockKey + "获得了锁");
                    return null;
                }
            }catch (Exception e){
                e.printStackTrace();
            }
            return null;
        }
    
        @Override
        public boolean release(String indentifier) {
            if(null != redLock){
                redLock.unlock();
                return true;
            }
    
            return false;
        }
    }
    复制代码

由于RedLock是针对主从和集群场景准备。上面代码采用哨兵模式。所以要让上面代码运行起来，需要先本地搭建Redis哨兵模式。

**具体测试代码如下所示：**

public class RedisDistributedRedLockTest {
        static int n = 5;
        public static void secskill() {
            if(n <= 0) {
                System.out.println("抢购完成");
                return;
            }
    
            System.out.println(--n);
        }
        public static void main(String[] args) {
    
            Config config = new Config();
            //支持单机，主从，哨兵，集群等模式
            //此为哨兵模式
            config.useSentinelServers()
                    .setMasterName("mymaster")
                    .addSentinelAddress("127.0.0.1:26369","127.0.0.1:26379","127.0.0.1:26389")
                    .setDatabase(0);
            Runnable runnable = () -> {
                RedisDistributedRedLock redisDistributedRedLock = null;
                RedissonClient redissonClient = null;
                try {
                    redissonClient = Redisson.create(config);
                    redisDistributedRedLock = new RedisDistributedRedLock(redissonClient, "stock_lock");
                    redisDistributedRedLock.acquire();
                    secskill();
                    System.out.println(Thread.currentThread().getName() + "正在运行");
                } finally {
                    if (redisDistributedRedLock != null) {
                        redisDistributedRedLock.release(null);
                    }
    
                    redissonClient.shutdown();
                }
            };
    
            for (int i = 0; i < 10; i++) {
                Thread t = new Thread(runnable);
                t.start();
            }
        }
    复制代码

# 总结 #

到此，基于Redis实现分布式锁的就告一段落了，由于分布式锁的实现方式主要有：数据库锁的方式、基于Redis实现和基于Zookeeper实现。

转载于:https://juejin.im/post/5cd3dec051882535b455aea8