2、分布式锁实现原理与最佳实践（二）

常见分布式锁的原理

4.1 Redisson

Redis 2.6之后才可以执行lua脚本，比起管道而言，这是原子性的，模拟一个商品减库存的原子操作：

//lua脚本命令执行方式：redis-cli --eval /tmp/test.lua , 10
jedis.set("product_stock_10016", "15");  
//初始化商品10016的库存
String script = " local count = redis.call('get', KEYS[1]) " +
        " local a = tonumber(count) " +
        " local b = tonumber(ARGV[1]) " +
        " if a >= b then " +
        "   redis.call('set', KEYS[1], a-b) " +
        "   return 1 " +
        " end " +
        " return 0 ";
Object obj = jedis.eval(script, Arrays.asList("product_stock_10016"), 
                        Arrays.asList("10"));
System.out.println(obj);

在这里插入图片描述
4.1.1 尝试加锁的逻辑

上面的org.redisson.RedissonLock#lock()通过调用自己方法内部的lock方法的org.redisson.RedissonLock#tryAcquire方法。之后调用 org.redisson.RedissonLock#tryAcquireAsync：

首先调用内部的org.redisson.RedissonLock#tryLockInnerAsync：设置对应的分布式锁
在这里插入图片描述
到这里获取锁的逻辑就结束了，如果这里没有获取到，在Future的回调里面就会直接return，会在外层有一个while true的循环，订阅释放锁的消息准备被唤醒。如果说加锁成功，就开始执行锁续命逻辑。

4.1.2 锁续命逻辑
lua脚本最后是以毫秒为单位返回key的剩余过期时间。成功加锁之后org.redisson.RedissonLock#scheduleExpirationRenewal中将会调用org.redisson.RedissonLock#renewExpiration，这个方法内部就有锁续命的逻辑，是一个定时任务，等10s执行。
执行的时候尝试执行的续命逻辑使用的是Lua脚本，当前的锁有值，就续命，没有就直接返回0：
在这里插入图片描述
返回0之后外层会判断，延时成功就会再次调用自己，否则延时调用结束，不再为当前的锁续命。所以这里的续命不是一个真正的定时，而是循环调用自己的延时任务。

4.1.3 循环间隔抢锁机制
如果一开始就加锁成功就直接返回。
如果一开始加锁失败，没抢到锁的线程就会在while循环中尝试加锁，加锁成功就结束循环，否则等待当前锁的超时时间之后再次尝试加锁。所以实现逻辑默认是非公平锁：
在这里插入图片描述
里面有一个subscribe的逻辑，会监听对应加锁的key，当锁释放之后publish对应的消息，此时如果没有到达对应的锁的超时时间，也会尝试获取锁，避免时间浪费。
4.1.4 释放锁和唤醒其他线程的逻辑
前面没有抢到锁的线程会监听对应的queue，后面抢到锁的线程释放锁的时候会发送一个消息。
在这里插入图片描述
订阅的时候指定收到消息时候的逻辑：会唤醒阻塞之后执行while循环

4.1.5 重入锁的逻辑
存在对应的锁，就对对应的hash结构的value直接+1，和Java重入锁的逻辑是一致的。
在这里插入图片描述

4.2 RedLock解决非单体项目的Redis主从架构的锁失效

https://redis.io/docs/manual/patterns/distributed-locks/
查看Redis官方文档，对于单节点的Redis ，使用setnx和lua del删除分布式锁是足够的，但是主从架构的场景下：锁先加在一个master节点上，默认是异步同步到从节点，此时master挂了会选择slave为master，此时又可以加锁，就会导致超卖。但是如果使用zookeeper来实现的话，由于zk是CP的，所以CP不存在这样的问题。
Redis文档中给出了RedLock的解决办法，使用redLock真的可以解决吗？
4.2.1 RedLock 原理
基于客户端的实现，是基于多个独立的Redis Master节点的一种实现（一般为5）。client依次向各个节点申请锁，若能从多数个节点中申请锁成功并满足一些条件限制，那么client就能获取锁成功。它通过独立的N个Master节点，避免了使用主备异步复制协议的缺陷，只要多数Redis节点正常就能正常工作，显著提升了分布式锁的安全性、可用性。
在这里插入图片描述
注意图中所有的节点都是master节点。加锁超过半数成功，就认为是成功。具体流程：
获取锁

获取当前时间T1，作为后续的计时依据；

按顺序地，依次向5个独立的节点来尝试获取锁 SET resource_name my_random_value NX PX 30000；

计算获取锁总共花了多少时间，判断获取锁成功与否；

时间：T2-T1；

多数节点的锁（N/2+1）；

当获取锁成功后的有效时间，要从初始的时间减去第三步算出来的消耗时间；

如果没能获取锁成功，尽快释放掉锁。

释放锁

向所有节点发起释放锁的操作，不管这些节点有没有成功设置过。

public String redlock() {
    String lockKey = "product_001";
    //这里需要自己实例化不同redis实例的redisson客户端连接，这里只是伪代码用一个redisson客户端简化了
    RLock lock1 = redisson.getLock(lockKey);
    RLock lock2 = redisson.getLock(lockKey);
    RLock lock3 = redisson.getLock(lockKey);

    /**
     * 根据多个 RLock 对象构建 RedissonRedLock （最核心的差别就在这里）
     */
    RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
    try {
        /**
         * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
         * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
         */
        boolean res = redLock.tryLock(10, 30, TimeUnit.SECONDS);
        if (res) {
            //成功获得锁，在这里处理业务
        }
    } catch (Exception e) {
        throw new RuntimeException("lock fail");
    } finally {
        //无论如何, 最后都要解锁
        redLock.unlock();
    }

    return "end";
}

但是，它的实现建立在一个不安全的系统模型上的，它依赖系统时间，当时钟发生跳跃时，也可能会出现安全性问题。分布式存储专家Martin对RedLock的分析文章，Redis作者的也专门写了一篇文章进行了反驳。

Martin Kleppmann：How to do distributed locking

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
Antirez：Is Redlock safe?

http://antirez.com/news/101
4.2.2 RedLock 问题一：持久化机制导致重复加锁
如果是上面的架构图，一般生产都不会配置AOF的每一条命令都落磁盘，一般会设置一些间隔时间，比如1s，如果ABC节点加锁成功，有一个节点C恰好是在1s内加锁，还没有落盘，此时挂了，就会导致其他客户端通过CDE又会加锁成功。
4.2.3 RedLock 问题二：主从下重复加锁
在这里插入图片描述
除非多部署一些节点，但是这样会导致加锁时间变长，这样比较下来效果就不如zk了。
4.2.4 RedLock 问题三：时钟跳跃导致重复加锁
C节点发生了时钟跳跃，导致加上的锁没有到达实际的超时时间，就被误以为超时而释放，此时其他客户端就可以重复加锁了。
4.3 Curator

InterProcessMutex 可重入锁的分析
在这里插入图片描述
五、业务中使用分布式锁的注意点

获取的锁要设置有效期，假设我们未设置key自动过期时间，在Set key value NX 后，如果程序crash或者发生网络分区后无法与Redis节点通信，毫无疑问其他 client 将永远无法获得锁，这将导致死锁，服务出现中断。
SETNX和EXPIRE命令去设置key和过期时间，这也是不正确的，因为你无法保证SETNX和EXPIRE命令的原子性。
自己使用 setnx 实现Redis锁的时候，注意并发情况下不要释放掉别人的锁（业务逻辑执行时间超过锁的过期时间），导致恶性循环。一般：
1）加锁的时候需要指定value的内容是当前进程中的当前线程的唯一标记，不要使用线程ID作为当前线程的锁的标记，因为不同实例上的线程ID可能是一样的。
2）释放锁的逻辑会写在finally ，释放锁时候要判断锁对应的value，而且要使用lua脚本实现原子 del 操作。因为if逻辑判断完之后也可能失效导致删除别人的锁。
3）针对扣减库存这个逻辑，lua脚本里面实现Redis比较库存、扣减库存操作的原子性。通过判断Redis Decr命令的返回值即可。此命令会返回扣减后的最新库存，若小于0则表示超卖。

5.1 自己实现分布式锁的坑

setnx不关心锁的顺序导致删除别人的锁
锁失效之后，别人加锁成功，自己把别人的锁删了。
我们无法预估程序执行需要的锁的时间。

public String deductStock() {
    String lockKey = "lock:product_101";
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "deltaqin");
    stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

    try {
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
        if (stock > 0) {
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
            System.out.println("扣减成功，剩余库存:" + realStock);
        } else {
            System.out.println("扣减失败，库存不足");
        }
    } finally {
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

setnx关心锁的顺序还是删除了别人的锁
并发会卡在各种地方，卡住的时候过期了，就会删掉别人加的锁：
错误的原因还是因为解锁的逻辑不是原子性的，这里可以参考Redisson的解锁逻辑使用lua脚本实现。

public String deductStock() {
    String lockKey = "lock:product_101";
    String clientId = UUID.randomUUID().toString();
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
    if (!result) {
        return "error_code";
    }
    try {
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
        if (stock > 0) {
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
            System.out.println("扣减成功，剩余库存:" + realStock);
        } else {
            System.out.println("扣减失败，库存不足");
        }
    } finally {
        if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
            // 卡在这里，锁过期了，其他线程又可以加锁，此时又把其他线程新加的锁删掉了
            stringRedisTemplate.delete(lockKey);
        }
    }
    return "end";
}

解决办法
这种问题解决的办法就是使用锁续命，比如使用一个定时任务间隔小于锁的超时时间，每隔一段时间就给锁续命，除非线程自己主动删除。这也是Redisson的实现思路。
5.2 锁优化：分段加锁逻辑

针对一个商品，要开启秒杀的时候，会将商品的库存预先加载到Redis缓存中，比如有100个库存，此时可以分为5个key，每一个key有20个库存。可以把分布式锁的性能提升5倍。
例如：

product_10111_stock = 100

product_10111_stock1 = 20

product_10111_stock2 = 20

product_10111_stock3 = 20

product_10111_stock4 = 20

product_10111_stock5 = 20