Redis实战（4）——Redisson分布式锁

1 基于互斥命令实现分布式锁的弊端

根据上篇文章基于redis互斥命令实现的分布式锁任然存在一定的弊端

1无法重入: 同一个线程无法重新获得同一把锁
2超时删除：会因为超时、任务阻塞而自动释放锁，出现其他线程抢占锁出现并行导致线程不安全的问题
3 不可重试：基于setnx 互斥指令实现的非阻塞式分布式锁在获取不到锁时将会立即返回，没有重试机制
4主从一致性：如果Redis提供了主从同步，主从同步时出现了延迟时，会出现无法判定当前线程锁的状态，出现线程不安全的问题。因为一般写指令【setnx】向主Redis操作，读指令【get key】向从Redis操作，即主从分离的情形。

2 Redisson

为了解决上述基于互斥命令实现的锁出现的问题，可以选择使用Redisson分布式锁方案。
Redisson不仅仅是一个Redis客户端，它还实现了很多具有分布式特性的常用工具类。
引入依赖

        <!--redisson 分布式锁-->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.5.6</version>
        </dependency>

创建Redisson的客户端

@Configuration
public class RedissonClientConfig {
    
    @Bean
    public RedissonClient redissonClient(){
        // 配置类
        Config config = new Config();
        // 添加redis地址，这里添加了单点的地址，也可以使用config.useClusterServers()添加集群地址
        config.useSingleServer()
                .setAddress("redis://127.0.0.1:6379")
                .setDatabase(1);
        // 创建客户端
        return Redisson.create(config);
    }
}

3 Redisson 分布式锁实现原理

3.1 Redisson实现可重入锁原理

相较于redis使用互斥指令setnex 创建的分布式锁，存储的数据结构为 $\textcolor{red}{String}$ ，Redisson存储的数据结构为 $\textcolor{red}{Hash}$ ，线程在尝试获得锁的时候,除了存储当前线程标识之外，还会存储锁的重入次数。
同一个线程获得锁时，重入次数加一。释放锁时，重入次数减一，直至减至0时redis数据库删除该key的信息。为了基于原子性操作，Redisson获得锁和释放锁的逻辑都是基于Lua脚本实现的

在这里插入图片描述
$\textcolor{red}{执行获得锁的底层代码执行Lua脚本}$

    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  //判定是否存在锁
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      //不存在，初次设置锁标识，重入次数为1
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      //设置锁的过期时间
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  //判定是否为当前线程id 持有锁
                  "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; " +
                  //未获得锁，返回当前锁的剩余的有效期时间 【单位ms】
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

KEYS[1] 锁的key
ARGV[1] 有效时长
ARGV[2] 锁的线程标识

$\textcolor{red}{释放锁的代码执行Lua脚本}$

    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                 //锁不存在时，发布锁释放的消息
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                "end;" +
                //不是当前线程标识，无法释放锁
                "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 " +
                    //重入次数未减至0 时，更新锁的持有时间
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                "else " +
                    //重入次数减至0时，redis删除锁，锁成功释放
                    "redis.call('del', KEYS[1]); " +
                    //发布锁成功释放的消息
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; "+
                "end; " +
                "return nil;",
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

    }

3.2 锁重试与锁的超时释放问题

在内部Redisson获得锁时，会执行以下方法。当成功获得锁时，返回null，获得锁失败时，获得锁剩余的持有时间。利用redis的订阅和信号量机制，在设定的等待时间内尝试重试获得锁。做到了锁重试，且不是无休止的盲目等待去获得锁的信息。
至于锁的超时释放问题，redisson 提供了watchdog机制，当不设定锁的超时时间，即默认设置为-1 时，利用watchdog机制，每隔一段时间 (internalLockLeaseTime 3)，重置锁的有效时长

在这里插入图片描述

$\textcolor{red}{锁在最大等待时间内进行锁重试}$

    //超时释放时间，时间单位，线程标识id
    <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        //超时时间转化为ms，存入内部成员变量中
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  //锁不存在的时候，设置锁
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
    }

    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        //设定的锁等待时间，转为ms 单位
        long time = unit.toMillis(waitTime);
        //当前时间
        long current = System.currentTimeMillis();
        final long threadId = Thread.currentThread().getId();
        //获得锁的剩余有效时间。根据上述方法值返回
        Long ttl = tryAcquire(leaseTime, unit, threadId);
        // ttl=null，标识锁获得成功，返回true
        if (ttl == null) {
            return true;
        }
        
        //更新锁等待时长，减去初次获得锁的时间
        time -= (System.currentTimeMillis() - current);
        if (time <= 0) {
            //等待时长<0，即已超出设定的超时等待时长。获得锁失败
            acquireFailed(threadId);
            return false;
        }
        
        current = System.currentTimeMillis();
        //订阅锁释放的信息
        final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        //在剩余等待时长中等待看是否有其它线程释放锁的信息，没有收到释放锁的信息
        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
            //超时，取消锁释放的订阅
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
                    @Override
                    public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
                        if (subscribeFuture.isSuccess()) {
                            unsubscribe(subscribeFuture, threadId);
                        }
                    }
                });
            }
            //判定没有收到锁释方的消息，获得锁失败
            acquireFailed(threadId);
            return false;
        }

        try {
            //再次更新锁的等待时间
            time -= (System.currentTimeMillis() - current);
            if (time <= 0) {
                //<0，即超时，获得锁失败
                acquireFailed(threadId);
                return false;
            }
        
            //进入重新获得锁逻辑
            while (true) {
                long currentTime = System.currentTimeMillis();
                //重新获得锁的剩余时间
                ttl = tryAcquire(leaseTime, unit, threadId);
                //null，成功获得锁
                if (ttl == null) {
                    return true;
                }

                //没有获得锁成功，更新锁的等待时间
                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                     //等待时间<0，即超时，获得锁失败
                    acquireFailed(threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= (System.currentTimeMillis() - currentTime);
                if (time <= 0) {
                    acquireFailed(threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }

$\textcolor{red}{锁超时——watchdog机制}$

    private RFuture<Boolean> tryAcquireOnceAsync(long leaseTime, TimeUnit unit, final long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        }
        //当设定超时时间为-1时
        RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        ttlRemainingFuture.addListener(new FutureListener<Boolean>() {
            @Override
            public void operationComplete(Future<Boolean> future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }
                Boolean ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining) {
                    //更新锁的有效时长 。
                    //scheduleExpirationRenewal是一个定时任务。任务的间隔时间是 internalLockLeaseTime / 3 。internalLockLeaseTime 设定的是30s，即锁的watchdog时间。直到用户显示的执行unlock()方法，取消该定时任务，锁成功释放。watchdog机制保证了锁的超时释放问题。
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

3.3 Redisson主从一致性解决方案

在redis 集群模式下，由于主从节点 $\textcolor{red}{读写分离}$ ，出现的因为主从同步出现延迟或主从同步数据失败会导致多个线程获得锁出现线程不安全问题。在Redisson中可以采用RedissonMultiLock【联锁，把多个锁联合成一把锁来看待】来解决。
即每一个redis节点都当成Master节点来看待，在获得锁时，必须每一个Redis节点都获得锁成功才算成功，释放锁时需要每一个Redis节点都释放锁成功才算成功。

$\textcolor{red}{MultiLock的使用}$

// 初始化三个锁 ,指向不同的redis节点
RLock lock1 = redissonClient.getLock("lockName1");
RLock lock2 = redissonClient.getLock("lockName2");
RLock lock3 = redissonClient.getLock("lockName3");

// 初始化三个锁的合并锁
RLock multiLock = redissonClient.getMultiLock(lock1, lock2, lock3);

// 获取锁
try{
multiLock.lock();
//do something
}finally{
 multiLock.unlock();
}