锐客网

  1. 首页 > 睿知 > it技术 > >

Springboot基于Redisson实现Redis分布式可重入锁源码解析

目录

  • 一、前言
  • 二、为什么使用Redisson
    • 1.我们打开官网
    • 2.我们可以看到官方让我们去使用其他
    • 3.打开官方推荐
    • 4.找到文档
  • 三、Springboot整合Redisson
    • 1.导入依赖
    • 2.以官网为例查看如何配置
    • 3.编写配置类
    • 4.官网测试加锁例子
    • 5.根据官网简单Controller接口编写
    • 6.测试
  • 四、lock.lock()源码分析
    • 1.打开RedissonLock实现类
    • 2.找到实现方法
    • 3.按住Ctrl进去lock方法
    • 4.进去尝试获取锁方法
    • 5.查看tryLockInnerAsync()方法
    • 6.进入4留下的定时任务scheduleExpirationRenewal()方法
  • 五、lock.lock(10,TimeUnit.SECONDS)源码分析
    • 六、lock.unlock()源码分析
      • 七、总结

        一、前言 我们在实现使用Redis实现分布式锁，最开始一般使用SET resource-name anystring NX EX max-lock-time进行加锁，使用Lua脚本保证原子性进行实现释放锁。这样手动实现比较麻烦，对此Redis官网也明确说Java版使用Redisson来实现。小编也是看了官网慢慢的摸索清楚，特写此记录一下。从官网到整合Springboot到源码解读，以单节点为例，小编的理解都在注释里，希望可以帮助到大家！！

        二、为什么使用Redisson
        1. 我们打开官网
        redis中文官网

        2. 我们可以看到官方让我们去使用其他
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        3. 打开官方推荐
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        4. 找到文档
        Redisson地址
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片

        【Springboot基于Redisson实现Redis分布式可重入锁源码解析】5. Redisson结构
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        三、Springboot整合Redisson
        1. 导入依赖
        org.springframework.bootspring-boot-starter-data-redisredis.clientsjedisorg.redissonredisson3.12.0


        2. 以官网为例查看如何配置
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        3. 编写配置类
        import org.redisson.Redisson; import org.redisson.api.RedissonClient; import org.redisson.config.Config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @author wangzhenjun * @date 2022/2/9 9:57 */@Configurationpublic class MyRedissonConfig {/*** 所有对redisson的使用都是通过RedissonClient来操作的* @return*/@Bean(destroyMethod="shutdown")public RedissonClient redisson(){// 1. 创建配置Config config = new Config(); // 一定要加redis://config.useSingleServer().setAddress("redis://192.168.17.130:6379"); // 2. 根据config创建出redissonClient实例RedissonClient redissonClient = Redisson.create(config); return redissonClient; }}


        4. 官网测试加锁例子
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        5. 根据官网简单Controller接口编写
        @ResponseBody@GetMapping("/hello")public String hello(){// 1.获取一把锁，只要锁名字一样，就是同一把锁RLock lock = redisson.getLock("my-lock"); // 2. 加锁lock.lock(); // 阻塞试等待默认加的都是30s// 带参数情况// lock.lock(10, TimeUnit.SECONDS); // 10s自动解锁，自动解锁时间一定要大于业务的执行时间。try {System.out.println("加锁成功" + Thread.currentThread().getId()); Thread.sleep(30000); } catch (InterruptedException e) {e.printStackTrace(); } finally {// 3. 解锁System.out.println("解锁成功：" + Thread.currentThread().getId()); lock.unlock(); }return "hello"; }


        6. 测试
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        四、lock.lock()源码分析
        1. 打开RedissonLock实现类
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片


        2. 找到实现方法
        @Overridepublic void lock() {try {// 我们发现不穿过期时间源码默认过期时间为-1lock(-1, null, false); } catch (InterruptedException e) {throw new IllegalStateException(); }}


        3. 按住Ctrl进去lock方法
        private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException { // 获取线程的id，占有锁的时候field的值为UUID:线程号idlong threadId = Thread.currentThread().getId(); // 尝试获得锁Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired 获得锁，返回if (ttl == null) {return; } // 这里说明获取锁失败，就通过线程id订阅这个锁RFuture future = subscribe(threadId); if (interruptibly) {commandExecutor.syncSubscriptionInterrupted(future); } else {commandExecutor.syncSubscription(future); }try {// 这里进行自旋，不断尝试获取锁while (true) {// 继续尝试获取锁ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired 获取成功if (ttl == null) {// 直接返回，挑出自旋break; }// waiting for message 继续等待获得锁if (ttl >= 0) {try {future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } catch (InterruptedException e) {if (interruptibly) {throw e; }future.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); }} else {if (interruptibly) {future.getNow().getLatch().acquire(); } else {future.getNow().getLatch().acquireUninterruptibly(); }}}} finally {// 取消订阅unsubscribe(future, threadId); }//get(lockAsync(leaseTime, unit)); }


        4. 进去尝试获取锁方法
        Springboot基于Redisson实现Redis分布式可重入锁源码解析
        文章图片

        private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { // 直接进入异步方法return get(tryAcquireAsync(leaseTime, unit, threadId)); }private RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {// 这里进行判断如果没有设置参数leaseTime = -1if (leaseTime != -1) {return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); }// 此方法进行获得锁，过期时间为看门狗的默认时间// private long lockWatchdogTimeout = 30 * 1000; 看门狗默认过期时间为30s// 加锁和过期时间要保证原子性，这个方法后面肯定调用执行了Lua脚本，我们下面在看RFuture ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); // 开启一个定时任务进行不断刷新过期时间ttlRemainingFuture.onComplete((ttlRemaining, e) -> {if (e != null) {return; }// lock acquired 获得锁if (ttlRemaining == null) {// 刷新过期时间方法，我们下一步详细说一下scheduleExpirationRenewal(threadId); }); return ttlRemainingFuture;


        5. 查看tryLockInnerAsync()方法
        RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {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; " +// hexists查看哈希表的指定字段是否存在，存在锁并且是当前线程持有锁"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +// hincrby自增一"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +// 锁的值大于1，说明是可重入锁，重置过期时间"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 锁已存在，且不是本线程，则返回过期时间ttl"return redis.call('pttl', KEYS[1]); ",Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); }

        6. 进入4留下的定时任务scheduleExpirationRenewal()方法
        一步步往下找源码：scheduleExpirationRenewal --->renewExpiration
        根据下面源码，定时任务刷新时间为：internalLockLeaseTime / 3，是看门狗的1/3，即为10s刷新一次
        private void renewExpiration() {ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ee == null) {return; }Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName()); if (ent == null) {return; }Long threadId = ent.getFirstThreadId(); if (threadId == null) {return; }RFuture future = renewExpirationAsync(threadId); future.onComplete((res, e) -> {if (e != null) {log.error("Can't update lock " + getName() + " expiration", e); return; }if (res) {// reschedule itselfrenewExpiration(); }}); }}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); ee.setTimeout(task); }


        五、lock.lock(10, TimeUnit.SECONDS)源码分析 1. 打开实现类
        @Overridepublic void lock(long leaseTime, TimeUnit unit) {try {// 这里的过期时间为我们输入的10lock(leaseTime, unit, false); } catch (InterruptedException e) {throw new IllegalStateException(); }}

        2. 方法lock()实现展示，同三.3源码
        3. 直接来到尝试获得锁tryAcquireAsync()方法
        private RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {// 这里进行判断如果没有设置参数leaseTime = -1,此时我们为10if (leaseTime != -1) {// 来到此方法return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); }// 此处省略后面内容，前面以详细说明。。。。}

        4. 打开tryLockInnerAsync()方法
        我们不难发现和没有传过期时间的方法一样，只不过leaseTime的值变了。
        RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {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; " +// hexists查看哈希表的指定字段是否存在，存在锁并且是当前线程持有锁"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +// hincrby自增一"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +// 锁的值大于1，说明是可重入锁，重置过期时间"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 锁已存在，且不是本线程，则返回过期时间ttl"return redis.call('pttl', KEYS[1]); ",Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); }

        六、lock.unlock()源码分析 1. 打开方法实现
        @Overridepublic void unlock() {try {// 点击进入释放锁方法get(unlockAsync(Thread.currentThread().getId())); } catch (RedisException e) {if (e.getCause() instanceof IllegalMonitorStateException) {throw (IllegalMonitorStateException) e.getCause(); } else {throw e; }}//Future future = unlockAsync(); //future.awaitUninterruptibly(); //if (future.isSuccess()) {//return; //}//if (future.cause() instanceof IllegalMonitorStateException) {//throw (IllegalMonitorStateException)future.cause(); //}//throw commandExecutor.convertException(future); }

        2. 打开unlockAsync()方法
        @Overridepublic RFuture unlockAsync(long threadId) {RPromise result = new RedissonPromise(); // 解锁方法，后面展开说RFuture future = unlockInnerAsync(threadId); // 完成future.onComplete((opStatus, e) -> {if (e != null) {// 取消到期续订cancelExpirationRenewal(threadId); // 将这个未来标记为失败并通知所有人result.tryFailure(e); return; }// 状态为空，说明解锁的线程和当前锁不是同一个线程if (opStatus == null) {IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "+ id + " thread-id: " + threadId); result.tryFailure(cause); return; }cancelExpirationRenewal(threadId); result.trySuccess(null); }); return result; }

        3. 打开unlockInnerAsync()方法
        protected RFuture unlockInnerAsync(long threadId) {return commandExecutor.evalWriteAsync(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); " +// 判断剩余数量是否大于0"if (counter > 0) then " +// 大于0 ，则刷新过期时间"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +"else " +// 释放锁，删除key并发布锁释放的消息"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; "+"end; " +"return nil; ",Arrays.asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)); }

        七、总结 这样大家就跟着小编走完了一遍底层源码，是不是感觉自己又行了，哈哈哈。小编走下来一遍觉得收货还是蛮大的，以前不敢点进去源码，进去就懵逼了，所以人要大胆的向前迈出第一步。
        到此这篇关于Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】的文章就介绍到这了,更多相关SpringbootRedis分布式可重入锁内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

          推荐阅读


          上一篇：Python中bytes字节串和string字符串之间的转换方法

          下一篇：python人工智能tensorflow构建卷积神经网络CNN