【Redis实战篇】分布式锁-Redisson

1. 分布式锁-redisson功能介绍

基于setnx实现的分布式锁存在下面的问题：

重入问题： 重入问题是指 获得锁的线程可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，比如HashTable这样的代码中，他的方法都是使用synchronized修饰的，假如他在一个方法内，调用另一个方法，那么此时如果是不可重入的，不就死锁了吗？所以可重入锁他的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

不可重试： 是指目前的分布式只能尝试一次，我们认为合理的情况是：当线程在获得锁失败后，他应该能再次尝试获得锁。

超时释放： 我们在加锁时增加了过期时间，这样我们可以防止死锁，超时时间太长也会影响业务的执行效率，但是如果卡顿的时间较短，业务还没执行完就释放了，也有安全隐患。

主从一致性： 如果Redis提供了主从集群，当我们向集群写数据时，主机需要异步的将数据同步给从机，而万一在同步过去之前，主机宕机了，就会出现死锁问题。

而这时我们就不得不引入我们的Redisson了，那么什么是Redisson呢？

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。
Redission提供了分布式锁的多种多样的功能。

在前面的【Redis】Java操作Redis之SpringDataRedis 章节我们也简单介绍了Redis的客户端其中之一Redisson

官网地址：https://redisson.org

2. redisson快速入门

引入依赖：

<dependency>
      <groupId>org.redisson</groupId>
      <artifactId>redisson</artifactId>
      <version>3.13.6</version>
 </dependency>

配置redisson客户端：

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.88.100:6379").setPassword("1234");
        //创建RedissonClient对象
        return Redisson.create(config);
    }
}

在 VoucherOrderServiceImpl注入RedissonClient

@Autowired
private RedissonClient redissonClient;

@Override
public Result seckillVoucher(Long voucherId) {
    //1.查询优惠券
    SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
    //2.判断秒杀活动是否开始
    if (voucher.getBeginTime().isAfter(LocalDateTime.now())) {
        return Result.fail("秒杀活动尚未开始!");
    }
    //3. 判断秒杀活动是否结束
    if (voucher.getEndTime().isBefore(LocalDateTime.now())) {
        return Result.fail("秒杀活动已结束!");
    }

    //4. 判断库存是否充足
    if (voucher.getStock() < 1) {
        return Result.fail("库存不足!");
    }

    Long userId = UserHolder.getUser().getId();

    //使用Redis自定义分布式锁解决集群环境下多进程不可见问题
    //SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
    //获取锁
    //boolean isLock = lock.tryLock(1200L);

    //使用Redisson分布式锁
    RLock lock = redissonClient.getLock("lock:order:" + userId);
    //获取锁 不加参数表示默认不重试, 超时释放时间为30s
    boolean isLock = lock.tryLock();

    //判断获取锁是否成功
    if (!isLock) {
        //获取锁失败
        return Result.fail("不允许重复下单操作!");
    }
    try {
        //获取代理对象
        IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
        return proxy.createVoucherOrder(voucherId);
    } finally {
        //释放锁
        lock.unlock();
    }

}

继续使用jmeter工具做200个线程的并发测试，然后查看数据库结果也能达到我们的预期，一个用户只能下一单。

3. redisson可重入锁原理

ReentrantLock可重入原理：

在java的ReentrantLock锁中，他是借助于底层的被voaltile修饰的一个state变量来记录重入的状态的，比如当前没有人持有这把锁，那么state=0，假如有人持有这把锁，那么state=1，如果持有这把锁的人再次持有这把锁，那么state就会+1 ，如果是对于synchronized而言，他在c语言代码中会有一个count，原理和state类似，也是重入一次就加一，释放一次就-1 ，直到减少成0时，表示当前这把锁没有被人持有。

在redission中，我们也支持可重入锁

在redisson分布式锁中，他采用hash结构用来获取锁，其中大key表示表示这把锁是否存在，用小key(或者说field)表示当前这把锁被哪个线程持有，value则表示重入多少次。而什么时候释放锁呢，只有当最后一次value--的操作为0之后，说明该线程已经在其他地方都执行完了有关锁的业务逻辑，这个时候当前线程就可以顺利的去释放锁了。

当然，我们也要保证获取锁与释放锁的原子性，所以我们也必须使用lua脚本来解决问题。

解释图片中获取锁的lua脚本:
redis.call('exists', key)== 0 : 判断锁是否存在，如果判断结果等于0，就表示当前这把锁不存在。
redis.call('hset', key, threadId, 1):向redis当中存储数据
redis.call('expire', key, releaseTime)：设置锁的有效期
如果当前这把锁存在，则第一个条件不满足，再判断
redis.call('hexists', key, threadId) == 1：此时需要通过key和filed判断当前这把锁是否属于当前线程
redis.call('hincrby',key, threadId, '1') ：表示锁的可重入次数 + 1
redis.call('expire', key, releaseTime)：然后再对其设置过期时间
代码走到最后一行，说明获取的锁不是自己，获取锁失败return 0

以下是释放锁的lua脚本：

RedissonLock中获取锁的Lua脚本源码：

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        this.internalLockLeaseTime = unit.toMillis(leaseTime);
        return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command, 
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                "return nil; " +
                "end;" +
                " if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                "redis.call('pexpire', KEYS[1], ARGV[1]);" +
                "return nil; " +
                "end; " +
                "return redis.call('pttl', KEYS[1]);"
                , Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
    }

RedissonLock中释放锁的Lua脚本源码：

  protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then" +
                " return nil;" +
                "end;" +
                " local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1);" +
                " if (counter > 0) then" +
                    " redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "else " +
                    "redis.call('del', KEYS[1]); " +
                    "redis.call('publish', KEYS[2], ARGV[1]);" +
                    " return 1; " +
                "end; " +
                "return nil;",
                Arrays.asList(this.getName(), this.getChannelName()), LockPubSub.UNLOCK_MESSAGE, this.internalLockLeaseTime, this.getLockName(threadId));
    }

4. redisson锁重试和WatchDog机制

说明：由于课程中已经说明了有关tryLock的源码解析以及其看门狗原理，所以笔者在这里给大家分析lock()方法的源码解析，希望大家在学习过程中，能够掌握更多的知识。
👇锁重试源码的整个流程：

抢锁过程中，获得当前线程，通过tryAcquire进行抢锁，该抢锁逻辑和之前Lua脚本逻辑相同

1、先判断当前这把锁是否存在，如果不存在，向redis当中存储hash锁，返回null

2、判断当前这把锁是否是属于当前线程，如果是，获取锁，重入次数 + 1，则返回null

所以如果返回是null，则代表着当前这哥们已经抢锁完毕，或者可重入完毕，但是如果以上两个条件都不满足，则进入到第三个条件，返回的是锁的失效时间。

下面绿色方框的逻辑其实就是获取锁失败时，利用信号量和PubSub消息订阅机制进行等待，等待被释放锁的消息来唤醒。如果超过一定时间就不会再等待重试了，如果等待被唤醒成功了，就会发现有个while(true) 再次进行tryAcquire进行抢锁。

👇因为trylock方法有重载方法，一个是带参数，一个是不带参数，如果带带参数传入的值是-1，如果传入参数，则leaseTime是他本身，所以如果传入了参数，此时leaseTime != -1 则会进去抢锁，抢锁的逻辑就是之前说的那三个逻辑。

如果是没有传入时间，则此时也会进行抢锁， 而且抢锁时间是默认看门狗机制时间commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()

ttlRemainingFuture.onComplete((ttlRemaining, e) 这段回调函数相当于对以上抢锁进行了监听，也就是说当上边抢锁完毕后，此方法会被调用，具体调用的逻辑就是去后台开启一个线程，进行续约逻辑，也就是看门狗线程。

renewExpiration方法此逻辑就是续约逻辑，注意看commandExecutor.getConnectionManager().newTimeout() 此方法
Method( new TimerTask() {},参数2 ，参数3 )，指的是：通过参数2，参数3 去描述什么时候去做参数1的事情，参数2的internalLockLeaseTime其实就是看门狗时间30s

所以当30 / 3s也就是10s之后，此时这个timerTask就触发了，他就去进行续约，把当前这把锁续约成30s，如果操作成功，那么此时就会递归调用自己，再重新设置一个timeTaskr()，于是再过10s后又再设置一个timerTask，如此往复，完成不停的续约。

那么大家可以想一想，假设我们的线程出现了宕机他还会续约吗？当然不会，因为没有人再去调用renewExpiration这个方法，所以等到时间之后自然就释放了。

5. redisson锁的MutiLock原理

为了提高redis的可用性，我们会搭建集群或者主从，现在以主从为例

此时我们去写命令，写在主机上， 主机会将数据同步给从机，但是假设在主机还没有来得及把数据写入到从机去的时候，此时主机宕机，哨兵会发现主机宕机，并且选举一个slave变成master，而此时新的master中实际上并没有锁信息，此时锁信息就已经丢掉了。

为了解决这个问题，redission提出来了MutiLock锁，使用这把锁咱们就不使用主从了，每个节点的地位都是一样的， 这把锁加锁的逻辑需要写入到每一个主丛节点上，只有所有的服务器都写入成功，此时才是加锁成功，假设现在某个节点挂了，那么他去获得锁的时候，只要有一个节点拿不到，都不能算是加锁成功，就保证了加锁的可靠性。

那么MutiLock 加锁原理是什么呢？笔者画了一幅图来说明

当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功， 那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试。