下一篇：Redis分布式锁原理（二）——Redisson分布式锁源码浅析

        虽然目前Redisson框架已经帮我们封装好了分布式锁的实现逻辑，我们可以直接像调用本地锁一样使用即可，但本文并不直接剖析Redisson源码，而是首先分析在分布式场景下实现redis分布式锁需要注意哪些问题，这样之后阅读Redisson源码也会变得更容易。

目录

一、什么场景会用到分布式锁？

二、redis为什么可以实现分布式锁？

三、实现redis分布式锁需要注意哪些问题？

---

一、什么场景会用到分布式锁？

        在分布式场景下，我们不同的业务功能放在不同的服务器上，而这些不同的业务可能会去操作同一个数据库资源，如果这时候大并发进来，就可能会出现同时操作共享资源的情况，为了避免这种情况发生，我们想到的是加锁，如果这时候是在各自服务器的代码实现上加本地锁能够解决这种问题吗？

        答案当然是不能，来分析一下上述场景：

        整个系统的业务1、业务2、业务3放在不同的服务器上构成了分布式场景，这三个业务底层都会去操作同一个数据库，现在对这三个业务的代码实现上加上本地锁，即synchronized、ReentrantLock等，此时100个并发请求进入这个系统，假设到达这三个服务器的请求量分别为30、30、40，由于三个业务是在不同的服务器上，所以对于它们而言使用本地锁锁住的不会是同一个对象，因此能进入业务1的请求只有1条，进入业务2的请求也有1条，同理进入业务3的请求也是1条，最终数据库会同时接收到3条访问请求，出现了同时操作共享资源的情况。

        分布式锁是怎么解决这种问题的呢？就好比在这三个业务外边放一把大锁，这把锁也就脱离了“本地”的概念，三个服务器都可以去这个公共的地方抢这把锁，谁抢到了这把锁谁就可以去执行业务操作数据库，其他业务只能等待。

二、redis为什么可以实现分布式锁？

        上面所提到的公共的地方可以用redis代替，也就是说所有的服务器都连接上同一个redis，redis是一个缓存数据库，我们通过往这个缓存中存取标记的手段达到锁的获取和释放的效果，说到这里就不得不引出这个关键的命令了：set NX,它是一个原子性的命令，它能保证如果缓存中如果没有这个key时才会对其进行设置，如果这个key已经存在了那么这个命令就会执行失败。

        下面可以用xshell连接虚拟机后进行进行一个简单的测试，需要提前在虚拟机中安装好redis。首先复制多份会话模拟多个用户去抢redis锁，在确保redis在虚拟中正常运行的前提下输入命令“docker exec -it redis redis-cli”并发送到全部会话，使所有的会话都进入到redis容器中，接着输入命令“set locktest 123 NX”并同时发送到全部会话，模拟同一时刻多个会话去抢同一把锁。结果如下，会话1抢占锁成功，而会话2和会话3抢锁失败，而打开redis可以看到的确只有一条锁记录插入成功。

        这样看来，redis的确是可以用来实现分布式锁，原理就是使用set nx命令进行设值，若返回成功则代表拿锁成功，返回失败则代表拿锁失败。

三、实现redis分布式锁需要注意哪些问题？

        首先来看这样一段代码。

    @Override
    public List<Map<String, Object>> findList() {
        //尝试获取分布式锁，步骤一
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", "齐天小圣");
        if (lock) {
            //加锁成功，执行业务，步骤二
            List<Map<String, Object>> resultList = findListByDB();
            //删除锁，步骤三
            redisTemplate.delete("lock");
            return resultList;
        } else {
            //加锁失败，自旋重试，即重新调用本方法。
            try {
                Thread.sleep(300);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return findList();
        }
    }

        乍一看用redis完成一个分布式锁的整体逻辑好像也不是很复杂，无非就是先去获取分布式锁，如果拿到锁了就去执行相应的业务，没有拿到锁就自旋重试。但其实上面这段代码漏洞百出，实际想要完成一个完美的分布式锁是很复杂的，我们需要注意以下问题：

        1、死锁问题：如果某服务器拿锁成功，执行到步骤二时突然宕机没能正常的执行步骤三释放锁，那么其他正在等待锁的服务器则永远也拿不到锁了，这就是死锁问题。

        解决这个问题很简单，只需要在拿锁之后给锁上一个超时时间即可，即使业务过程中出现问题导致不能释放锁，到了过期时间redis也会自动帮我们去把这个锁删除。需要注意的是，千万不能把“获取锁”和“设置超时时间”在代码中分成两步执行，如下：

        原因在于这两个步骤分开执行没有保证原子性，拿锁到设置过期时间之间是存在时间差的，如果在这之间机器宕机了还是会存在上述问题，解决办法就是在占锁的同时设置过期时间。

        2、业务时间 > 超时时间：假设这样一个场景，业务1拿锁成功并设置过期时间30s，但业务1比较复杂需要花费40s，那么到了30s后业务1的锁就已经失效了，此时业务2抢到了锁也进来执行相应的逻辑，那么此时业务1和业务2都在执行各自的业务逻辑，可能会操作相同的数据资源造成违反资源互斥的现象。问题还没完，又过去了10s，业务1执行完了，理所当然的就去删锁，那么自然就会把业务2手里的锁删掉，业务2一脸懵逼。。。这就造成了锁误删的现象。

        先解决锁误删的问题，其实很简单，只需要保证谁拿的锁谁就有资格删就可以了，我们在获取锁的时候设置了一个value，此时就派上了用场，把每个业务的value都设置uuid，最后删锁的时候先去redis获取锁对应的值，如果这个值等于uuid才有资格执行删锁命令。需要注意的是这里也需要保证原子性，因为去远程redis获取锁对应的值再返回来也是有时间差的，如果业务1去远程获取到锁的value为“1111”，回来的过程中锁过期了，此时业务2拿到了锁开心的去执行它的业务去了，好景不长，业务1回来判断出锁的value是等于自己的uuid，于是又理所当然的把锁删掉了，业务2又一脸懵逼。。。。这里就需要使用lua脚本解锁，lua脚本就是为了保证这两段操作的原子性,解锁脚本内容如下：

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

        最后优化后的代码如下：

    @Override
    public List<Map<String, Object>> findList() {
        //尝试获取分布式锁,并设置过期时间
        String uuid = UUID.randomUUID().toString();
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 30, TimeUnit.SECONDS);
        if (lock) {
            //加锁成功，执行业务
            List<Map<String, Object>> resultList;
            try{
                resultList = findListByDB();
            } finally {
                //获取值进行对比，若对比成功则有资格进行删锁，需使用lua脚本保证原子性
                String script = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                        "then\n" +
                        "    return redis.call(\"del\",KEYS[1])\n" +
                        "else\n" +
                        "    return 0\n" +
                        "end";
                Long result = redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class)
                        , Arrays.asList("lock"), uuid);
            }

            return resultList;
        } else {
            //加锁失败，自旋重试，即重新调用本方法。
            try {
                Thread.sleep(300);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return findList();
        }
    }

        其实我们还遗漏了一个问题，上面优化后的代码并没有解决业务时间 > 超时时间造成的违反资源互斥的现象，解决这个问题的办法是如果业务还没有结束，那么这期间每隔一段时间就给锁进行延时，即“锁续命”，但这涉及到定时任务，异步编排，所以就不再往下继续优化，到了这里不得不请Redisson分布式锁隆重登场了。

        在实际开发中我们并不会像上面这样用原生redis代码去实现分布式锁，而是使用已经封装好的框架——Redisson，Redisson已经为我们解决了上述细节问题，包括用定时任务实现“看门狗”机制为锁延时，我们只需要像平时使用本地锁一样进行调用即可。

        有了上述的思考，我们在下篇再来探析Redisson源码就会变得更加明朗了。

下一篇：Redis分布式锁原理（二）——Redisson分布式锁源码浅析