Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】

简介: Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】

Springboot基于Redisson实现Redis分布式可重入锁【案例到源码分析】


一、前言

我们在实现使用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));
}

七、总结

这样大家就跟着小编走完了一遍底层源码,是不是感觉自己又行了,哈哈哈。小编走下来一遍觉得收货还是蛮大的,以前不敢点进去源码,进去就懵逼了,所以人要大胆的向前迈出第一步。一起加油吧,看到这里不一键三连,有点对不起小编了哦!!

相关文章
|
5月前
|
NoSQL Java 网络安全
SpringBoot启动时连接Redis报错:ERR This instance has cluster support disabled - 如何解决?
通过以上步骤一般可以解决由于配置不匹配造成的连接错误。在调试问题时,一定要确保服务端和客户端的Redis配置保持同步一致。这能够确保SpringBoot应用顺利连接到正确配置的Redis服务,无论是单机模式还是集群模式。
503 5
|
12月前
|
NoSQL Java 关系型数据库
微服务——SpringBoot使用归纳——Spring Boot 中集成Redis——Redis 介绍
本文介绍在 Spring Boot 中集成 Redis 的方法。Redis 是一种支持多种数据结构的非关系型数据库(NoSQL),具备高并发、高性能和灵活扩展的特点,适用于缓存、实时数据分析等场景。其数据以键值对形式存储,支持字符串、哈希、列表、集合等类型。通过将 Redis 与 Mysql 集群结合使用,可实现数据同步,提升系统稳定性。例如,在网站架构中优先从 Redis 获取数据,故障时回退至 Mysql,确保服务不中断。
426 0
微服务——SpringBoot使用归纳——Spring Boot 中集成Redis——Redis 介绍
|
6月前
|
NoSQL Java 调度
分布式锁与分布式锁使用 Redis 和 Spring Boot 进行调度锁(不带 ShedLock)
分布式锁是分布式系统中用于同步多节点访问共享资源的机制,防止并发操作带来的冲突。本文介绍了基于Spring Boot和Redis实现分布式锁的技术方案,涵盖锁的获取与释放、Redis配置、服务调度及多实例运行等内容,通过Docker Compose搭建环境,验证了锁的有效性与互斥特性。
545 0
分布式锁与分布式锁使用 Redis 和 Spring Boot 进行调度锁(不带 ShedLock)
|
缓存 NoSQL Java
基于SpringBoot的Redis开发实战教程
Redis在Spring Boot中的应用非常广泛,其高性能和灵活性使其成为构建高效分布式系统的理想选择。通过深入理解本文的内容,您可以更好地利用Redis的特性,为应用程序提供高效的缓存和消息处理能力。
1174 79
|
NoSQL Java Redis
Springboot使用Redis实现分布式锁
通过这些步骤和示例,您可以系统地了解如何在Spring Boot中使用Redis实现分布式锁,并在实际项目中应用。希望这些内容对您的学习和工作有所帮助。
1308 83
|
10月前
|
监控 Java 调度
SpringBoot中@Scheduled和Quartz的区别是什么?分布式定时任务框架选型实战
本文对比分析了SpringBoot中的`@Scheduled`与Quartz定时任务框架。`@Scheduled`轻量易用,适合单机简单场景,但存在多实例重复执行、无持久化等缺陷;Quartz功能强大,支持分布式调度、任务持久化、动态调整和失败重试,适用于复杂企业级需求。文章通过特性对比、代码示例及常见问题解答,帮助开发者理解两者差异,合理选择方案。记住口诀:单机简单用注解,多节点上Quartz;若是任务要可靠,持久化配置不能少。
920 4
|
存储 Java 文件存储
🗄️Spring Boot 3 整合 MinIO 实现分布式文件存储
本文介绍了如何基于Spring Boot 3和MinIO实现分布式文件存储。随着应用规模扩大,传统的单机文件存储方案难以应对大规模数据和高并发访问,分布式文件存储系统成为更好的选择。文章详细讲解了MinIO的安装、配置及与Spring Boot的整合步骤,包括Docker部署、MinIO控制台操作、Spring Boot项目中的依赖引入、配置类编写及工具类封装等内容。最后通过一个上传头像的接口示例展示了具体的开发和测试过程,强调了将API操作封装成通用工具类以提高代码复用性和可维护性的重要性。
2479 7
🗄️Spring Boot 3 整合 MinIO 实现分布式文件存储
|
9月前
|
机器学习/深度学习 数据采集 人机交互
springboot+redis互联网医院智能导诊系统源码,基于医疗大模型、知识图谱、人机交互方式实现
智能导诊系统基于医疗大模型、知识图谱与人机交互技术,解决患者“知症不知病”“挂错号”等问题。通过多模态交互(语音、文字、图片等)收集病情信息,结合医学知识图谱和深度推理,实现精准的科室推荐和分级诊疗引导。系统支持基于规则模板和数据模型两种开发原理:前者依赖人工设定症状-科室规则,后者通过机器学习或深度学习分析问诊数据。其特点包括快速病情收集、智能病症关联推理、最佳就医推荐、分级导流以及与院内平台联动,提升患者就诊效率和服务体验。技术架构采用 SpringBoot+Redis+MyBatis Plus+MySQL+RocketMQ,确保高效稳定运行。
664 0
|
12月前
|
存储 人工智能 NoSQL
SpringBoot整合Redis、ApacheSolr和SpringSession
本文介绍了如何使用SpringBoot整合Redis、ApacheSolr和SpringSession。SpringBoot以其便捷的配置方式受到开发者青睐,通过引入对应的starter依赖,可轻松实现功能整合。对于Redis,可通过配置RedisSentinel实现高可用;SpringSession则提供集群Session管理,支持多种存储方式如Redis;整合ApacheSolr时,借助Zookeeper搭建SolrCloud提高可用性。文中详细说明了各组件的配置步骤与代码示例,方便开发者快速上手。
222 11
|
12月前
|
NoSQL Java API
微服务——SpringBoot使用归纳——Spring Boot 中集成Redis——Spring Boot 集成 Redis
本文介绍了在Spring Boot中集成Redis的方法,包括依赖导入、Redis配置及常用API的使用。通过导入`spring-boot-starter-data-redis`依赖和配置`application.yml`文件,可轻松实现Redis集成。文中详细讲解了StringRedisTemplate的使用,适用于字符串操作,并结合FastJSON将实体类转换为JSON存储。还展示了Redis的string、hash和list类型的操作示例。最后总结了Redis在缓存和高并发场景中的应用价值,并提供课程源代码下载链接。
2611 0

热门文章

最新文章