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. 我们可以看到官方让我们去使用其他

3. 打开官方推荐

4. 找到文档

Redisson地址

5. Redisson结构

三、Springboot整合Redisson

1. 导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
</dependency>
<!--redis分布式锁-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.12.0</version>
</dependency>

2. 以官网为例查看如何配置

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
 */
@Configuration
public 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. 官网测试加锁例子

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. 测试

四、lock.lock()源码分析

1. 打开RedissonLock实现类

2. 找到实现方法

@Override
public void lock() {
    try {
    	// 我们发现不穿过期时间源码默认过期时间为-1
        lock(-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:线程号id
    long threadId = Thread.currentThread().getId();
    // 尝试获得锁
    Long ttl = tryAcquire(leaseTime, unit, threadId);
    // lock acquired 获得锁,返回
    if (ttl == null) {
        return;
    }
	// 这里说明获取锁失败,就通过线程id订阅这个锁
    RFuture<RedissonLockEntry> 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. 进去尝试获取锁方法

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {
	// 直接进入异步方法
    return get(tryAcquireAsync(leaseTime, unit, threadId));
}

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    // 这里进行判断如果没有设置参数leaseTime = -1
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 此方法进行获得锁,过期时间为看门狗的默认时间
    // private long lockWatchdogTimeout = 30 * 1000;看门狗默认过期时间为30s
    // 加锁和过期时间要保证原子性,这个方法后面肯定调用执行了Lua脚本,我们下面在看
    RFuture<Long> 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()方法

<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 " +
              		// 存在则获取锁
                  "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.<Object>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() {
        @Override
        public 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<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> {
                if (e != null) {
                    log.error("Can't update lock " + getName() + " expiration", e);
                    return;
                }

                if (res) {
                    // reschedule itself
                    renewExpiration();
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    ee.setTimeout(task);
}

五、lock.lock(10, TimeUnit.SECONDS)源码分析

1. 打开实现类

@Override
public void lock(long leaseTime, TimeUnit unit) {
    try {
    	// 这里的过期时间为我们输入的10
        lock(leaseTime, unit, false);
    } catch (InterruptedException e) {
        throw new IllegalStateException();
    }
}

2. 方法lock()实现展示,同三.3源码

3. 直接来到尝试获得锁tryAcquireAsync()方法

private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
    // 这里进行判断如果没有设置参数leaseTime = -1,此时我们为10
    if (leaseTime != -1) {
    	// 来到此方法
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 此处省略后面内容,前面以详细说明。。。。
}

4. 打开tryLockInnerAsync()方法

我们不难发现和没有传过期时间的方法一样,只不过leaseTime的值变了。

<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 " +
              		// 存在则获取锁
                  "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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

六、lock.unlock()源码分析

1. 打开方法实现

@Override
public void unlock() {
    try {
    	// 点击进入释放锁方法
        get(unlockAsync(Thread.currentThread().getId()));
    } catch (RedisException e) {
        if (e.getCause() instanceof IllegalMonitorStateException) {
            throw (IllegalMonitorStateException) e.getCause();
        } else {
            throw e;
        }
    }

//        Future<Void> future = unlockAsync();
//        future.awaitUninterruptibly();
//        if (future.isSuccess()) {
//            return;
//        }
//        if (future.cause() instanceof IllegalMonitorStateException) {
//            throw (IllegalMonitorStateException)future.cause();
//        }
//        throw commandExecutor.convertException(future);
}

2. 打开unlockAsync()方法

@Override
public RFuture<Void> unlockAsync(long threadId) {
    RPromise<Void> result = new RedissonPromise<Void>();
    // 解锁方法,后面展开说
    RFuture<Boolean> 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<Boolean> 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.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));

}

七、总结

这样大家就跟着小编走完了一遍底层源码,是不是感觉自己又行了,哈哈哈。小编走下来一遍觉得收货还是蛮大的,以前不敢点进去源码,进去就懵逼了,所以人要大胆的向前迈出第一步。

到此这篇关于Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】的文章就介绍到这了,更多相关SpringbootRedis分布式可重入锁内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • SpringBoot使用Redisson实现分布式锁(秒杀系统)

    前面讲完了Redis的分布式锁的实现,接下来讲Redisson的分布式锁的实现,一般提及到Redis的分布式锁我们更多的使用的是Redisson的分布式锁,Redis的官方也是建议我们这样去做的.Redisson点我可以直接跳转到Redisson的官方文档. 1.1.引入Maven依赖 <dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter&l

  • SpringBoot集成Redisson实现分布式锁的方法示例

    上篇 <SpringBoot 集成 redis 分布式锁优化>对死锁的问题进行了优化,今天介绍的是 redis 官方推荐使用的 Redisson ,Redisson 架设在 redis 基础上的 Java 驻内存数据网格(In-Memory Data Grid),基于NIO的 Netty 框架上,利用了 redis 键值数据库.功能非常强大,解决了很多分布式架构中的问题. Github的wiki地址: https://github.com/redisson/redisson/wiki 官方文档

  • Springboot中如何使用Redisson实现分布式锁浅析

    目录 前言 1. 概述 2. Redisson 在 Springboot 中的使用 2.1 引入依赖 2.2 在 Springboot 配置中配置Redis 2.3 Demo代码 3. 综述 前言 在分布式场景下为了保证数据最终一致性.在单进程的系统中,存在多个线程可以同时改变某个变量(可变共享变量)时,就需要对变量或代码块做同步(lock-synchronized),使其在修改这种变量时能够线性执行消除并发修改变量.但分布式系统是多部署.多进程的,开发语言提供的并发处理API在此场景下就无能为

  • SpringBoot整合Redisson实现分布式锁

    目录 一.添加依赖 二.redis配置文件 三.新建配置类 四.使用分布式锁 可重入锁 读写锁 信号量(Semaphore) 闭锁(CountDownLatch) Redisson是架设在redis基础上的一个Java驻内存数据网格(In-Memory Data Grid).充分的利用了Redis键值数据库提供的一系列优势,基于Java实用工具包中常用接口,为使用者提供了一系列具有分布式特性的常用工具类.使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力,大大降低

  • 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.查看tryLockInne

  • Netty分布式ByteBuf的分类方式源码解析

    目录 ByteBuf根据不同的分类方式 会有不同的分类结果 1.Pooled和Unpooled 2.基于直接内存的ByteBuf和基于堆内存的ByteBuf 3.safe和unsafe 上一小节简单介绍了AbstractByteBuf这个抽象类, 这一小节对其子类的分类做一个简单的介绍 ByteBuf根据不同的分类方式 会有不同的分类结果 我们首先看第一种分类方式 1.Pooled和Unpooled pooled是从一块内存里去取一段连续内存封装成byteBuf 具体标志是类名以Pooled开头

  • Redisson分布式锁源码解析

    Redisson锁继承Implements Reentrant Lock,所以具备 Reentrant Lock 锁中的一些特性:超时,重试,可中断等.加上Redisson中Redis具备分布式的特性,所以非常适合用来做Java中的分布式锁. 下面我们对其加锁.解锁过程中的源码细节进行一一分析. 锁的接口定义了一下方法: 分布式锁当中加锁,我们常用的加锁接口: boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws

  • Golang实现可重入锁的示例代码

    目录 什么是可重入锁 具体实现 项目中遇到了可重入锁的需求和实现,具体记录下. 什么是可重入锁 我们平时说的分布式锁,一般指的是在不同服务器上的多个线程中,只有一个线程能抢到一个锁,从而执行一个任务.而我们使用锁就是保证一个任务只能由一个线程来完成.所以我们一般是使用这样的三段式逻辑: Lock();DoJob();Unlock(); 但是由于我们的系统都是分布式的,这个锁一般不会只放在某个进程中,我们会借用第三方存储,比如 Redis 来做这种分布式锁.但是一旦借助了第三方存储,我们就必须面对

  • redis分布式锁之可重入锁的实现代码

    上篇redis实现的分布式锁,有一个问题,它不可重入. 所谓不可重入锁,即若当前线程执行某个方法已经获取了该锁,那么在方法中尝试再次获取锁时,就会获取不到被阻塞. 同一个人拿一个锁 ,只能拿一次不能同时拿2次. 1.什么是可重入锁?它有什么作用? 可重入锁,也叫做递归锁,指的是在同一线程内,外层函数获得锁之后,内层递归函数仍然可以获取到该锁. 说白了就是同一个线程再次进入同样代码时,可以再次拿到该锁. 它的作用是:防止在同一线程中多次获取锁而导致死锁发生. 2.那么java中谁实现了可重入锁了?

  • Redisson实现Redis分布式锁的几种方式

    目录 Redis几种架构 普通分布式锁 单机模式 哨兵模式 集群模式 总结 Redlock分布式锁 实现原理 问题合集 前几天发的一篇文章<Redlock:Redis分布式锁最牛逼的实现>,引起了一些同学的讨论,也有一些同学提出了一些疑问,这是好事儿.本文在讲解如何使用Redisson实现Redis普通分布式锁,以及Redlock算法分布式锁的几种方式的同时,也附带解答这些同学的一些疑问. Redis几种架构 Redis发展到现在,几种常见的部署架构有: 单机模式: 主从模式: 哨兵模式: 集

  • Java源码解析之可重入锁ReentrantLock

    本文基于jdk1.8进行分析. ReentrantLock是一个可重入锁,在ConcurrentHashMap中使用了ReentrantLock. 首先看一下源码中对ReentrantLock的介绍.如下图.ReentrantLock是一个可重入的排他锁,它和synchronized的方法和代码有着相同的行为和语义,但有更多的功能.ReentrantLock是被最后一个成功lock锁并且还没有unlock的线程拥有着.如果锁没有被别的线程拥有,那么一个线程调用lock方法,就会成功获取锁并返回.

  • java高并发的ReentrantLock重入锁

    目录 synchronized的局限性 ReentrantLock ReentrantLock基本使用 ReentrantLock是可重入锁 ReentrantLock实现公平锁 ReentrantLock获取锁的过程是可中断的 tryLock无参方法 tryLock有参方法 ReentrantLock其他常用的方法 获取锁的4种方法对比 总结 synchronized的局限性 synchronized是java内置的关键字,它提供了一种独占的加锁方式.synchronized的获取和释放锁由j

  • Netty分布式ByteBuf缓冲区分配器源码解析

    目录 缓冲区分配器 以其中的分配ByteBuf的方法为例, 对其做简单的介绍 跟到directBuffer()方法中 我们回到缓冲区分配的方法 然后通过validate方法进行参数验证 缓冲区分配器 顾明思议就是分配缓冲区的工具, 在netty中, 缓冲区分配器的顶级抽象是接口ByteBufAllocator, 里面定义了有关缓冲区分配的相关api 抽象类AbstractByteBufAllocator实现了ByteBufAllocator接口, 并且实现了其大部分功能 和AbstractByt

  • Netty分布式ByteBuf使用的底层实现方式源码解析

    目录 概述 AbstractByteBuf属性和构造方法 首先看这个类的属性和构造方法 我们看几个最简单的方法 我们重点关注第二个校验方法ensureWritable(length) 我们跟到扩容的方法里面去 最后将写指针后移length个字节 概述 熟悉Nio的小伙伴应该对jdk底层byteBuffer不会陌生, 也就是字节缓冲区, 主要用于对网络底层io进行读写, 当channel中有数据时, 将channel中的数据读取到字节缓冲区, 当要往对方写数据的时候, 将字节缓冲区的数据写到cha

随机推荐