【Redis】6、Redisson 分布式锁的简单使用(可重入、重试机制...)

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 【Redis】6、Redisson 分布式锁的简单使用(可重入、重试机制...)


零、自己通过 set nx ex 实现的分布式锁存在的问题

✏️ 不可重入 同一个线程无法多次获取同一把锁

✏️ 不可重试 获取锁只尝试一次就返回 false,没有重试机制

✏️ 超时释放 锁超时释放虽然可以避免死锁,但如果是业务执行耗时较长,也会导致锁释放,存在安全隐患

一、Redisson 介绍

✏️ Redisson 是一个在 Redis 基础上实现的 Java 驻内存数据网格(In-Memory Data Grid)

✏️ 它提供了一系列分布式的 Java 常用对象

✏️ 提供了许多分布式服务,其中包含了各种分布式锁的实现

官网地址 https://redisson.org

GitHub 地址 https://github.com/redisson/redisson

二、Redisson 基本使用(改造业务)

(1) 依赖

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

(2) 配置 Redisson 客户端

@Configuration
@SuppressWarnings("all")
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://192.168.88.130:6379")
                .setPassword("root");
        return Redisson.create(config);
    }
}

✏️ 配置 redis 地址也可以使用 config.useClusterServers() 添加集群地址

✏️ useSingleServer() 是添加单点地址

(3) 使用 Redisson 的可重入锁

@Test
 public void tesRedisson() throws InterruptedException {
     // 获取可重入锁, 指定锁的名称
     RLock lock = redissonClient.getLock("anLock");
     // 尝试获取锁
     // 参数1:获取锁的最大等待时间(期间会多次重试获取锁)
     // 参数2:锁自动释放时间
     // 参数3:时间单位
     boolean isGetLock = lock.tryLock(1, 10, TimeUnit.SECONDS);
     if (isGetLock) {
         try {
             System.out.println("执行业务");
         } finally {
             lock.unlock();
         }
     }
 }

三、Redisson 可重入锁原理

✏️ 通过 Redis 的 Hash 数据结构实现

✏️ 存入线程标识和 counter【记录锁被重入的次数(被使用的次数),被使用(获取)一次就递增1,被释放则 counter 减1】

✏️ 当 counter 为 0 的时候,整个锁会被释放

@Resource
    private RedissonClient redissonClient;
    @SuppressWarnings("all")
    @Test
    public void tRedisson() {
        RLock lock = redissonClient.getLock("lock");
        boolean getLock = lock.tryLock();
        if (!getLock) {
            System.out.println("tRedisson getLock Failed");
            return;
        }
        try {
            System.out.println("tRedisson getLock Success");
            m(lock);
        } finally {
            System.out.println("tRedisson unlock");
            lock.unlock();
        }
    }
    public void m(RLock lock) {
        boolean getLock = lock.tryLock();
        if (!getLock) {
            System.out.println("m() getLock Failed");
            return;
        }
        try {
            System.out.println("m() getLock Success");
        } finally {
            System.out.println("m() unlock");
            lock.unlock();
        }
    }


✏️ 没有直接使用 hset

✏️ hincrby 命令:当 key 不存在的时候会自动创建 key

✏️ 获取锁成功:返回 nil;获取锁失败:返回当前锁的存活时间

✏️ pttl毫秒为单位返回某个 key 的存活时间

✏️ redis.call('publish', KEYS[2], ARGV[1]); 发布释放锁的消息通知

四、Redisson 可重试原理

下面是 Redisson 的 tryLock() 方法的源码:

@Override
 public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
     // 把超时等待时间转换为毫秒
     long time = unit.toMillis(waitTime);
     // 获取当前时间(毫秒)
     long current = System.currentTimeMillis();
     // 获取线程 ID
     long threadId = Thread.currentThread().getId();
     // tryAcquire 尝试获取锁(获取锁成功返回 null, 获取锁失败返回剩余ttl)
     Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
     // lock acquired
     if (ttl == null) { // 成功获取锁
         return true;
     }
     
     // time 是超时释放时间(单位:毫秒)
     // time = time - (System.currentTimeMillis() - current)
     time -= System.currentTimeMillis() - current; // 刷新超时释放时间
     if (time <= 0) { // 超时释放时间到了
         acquireFailed(waitTime, unit, threadId);
         return false; // 获取锁失败
     }
     
     // 获取 当前时间毫秒值
     current = System.currentTimeMillis();
     // 等待释放锁的通知(订阅释放锁的信号)
     RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
  // 等到超时释放时间结束还没有收到释放锁的通知的话, 返回 false
     // 获取锁失败 
     if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
         if (!subscribeFuture.cancel(false)) { // 取消订阅
             subscribeFuture.onComplete((res, e) -> {
                 if (e == null) {
                     unsubscribe(subscribeFuture, threadId);
                 }
             });
         }
     
         acquireFailed(waitTime, unit, threadId);
         return false;
     }
     
     // 接收到释放锁的信号
     try {
      // 判断释放到超时时间
         time -= System.currentTimeMillis() - current;
         if (time <= 0) {
             acquireFailed(waitTime, unit, threadId);
             return false; // 获取锁失败
         }
     
         while (true) { // 反复尝试
             long currentTime = System.currentTimeMillis();
             ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
             // lock acquired
             if (ttl == null) { // 获取锁成功
                 return true;
             }
             time -= System.currentTimeMillis() - currentTime;
             if (time <= 0) {
                 acquireFailed(waitTime, unit, threadId);
                 return false;
             }
             // waiting for message
             // 代码来到这里:没有获取到锁, 超时时间有剩余
             // 等待释放锁的信号
             currentTime = System.currentTimeMillis();
             if (ttl >= 0 && ttl < time) {
                 subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
             } else {
                 subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
             }
             time -= System.currentTimeMillis() - currentTime;
             if (time <= 0) {
                 acquireFailed(waitTime, unit, threadId);
                 return false;
             }
         }
     } finally {
         unsubscribe(subscribeFuture, threadId);
     }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
 }
private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
      // 判断超时释放时间是否是 -1
      if (leaseTime != -1) {
          return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
      }
      // 代码来到这里表示超时释放时间是 -1
      // tryLockInnerAsync 异步的方法
      RFuture<Boolean> ttlRemainingFuture = tryLockInnerAsync(waitTime,
                                                  commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),
                                                  TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
      ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
          if (e != null) {
              return;
          }
          // lock acquired
          if (ttlRemaining) {
              scheduleExpirationRenewal(threadId);
          }
      });
      return ttlRemainingFuture;
  }

tryLockInnerAsync()、tryAcquire(): ① 尝试获取锁。获取锁成功,返回 nil;获取锁失败,返回超时是否时间(lessTime)

② 该方法是异步的

① 获取锁失败后,不会立刻重新尝试获取锁

② 会等待释放锁的通知

五、Redisson 超时释放(锁的 ttl)

好😵懵逼啊,随便看看得了

📝 可重入:利用hash结构记录线程id和重入次数

📝 可重试:利用信号量和 PubSub 功能实现等待、唤醒,获取锁失败的重试机制

📝 超时续约:利用 watchDog,每隔一段时间(releaseTime / 3),重置超时时间

六、主从一致(连锁 MultiLock)

✏️主节点支持写请求

✏️从节点支持读请求

✏️主节点写入的数据要同步到从节点

✏️假如向主节点写入了数据,主节点还没有来得及向从节点同步数据自己就宕机了,此时 Redis 的哨兵机制会从剩下的从节点中选一个从节点作为新的主节点

✏️ 该新的主节点是没有上个主节点中的 lock = thread1 数据的

✏️ 锁失效,可能存在线程安全问题


✏️其中有任何一个节点宕机都无法拿到锁

七、锁总结

🎄不可重入 Redis 分布式锁:

原理: 利用 setnx 的互斥性;利用 ex 避免死锁;释放锁时判断线程标

缺陷: 不可重入、无法重试、锁超时失效

🎄可重入的 Redis 分布式锁:

原理: 利用 Hash 结构,记录线程标识和重入次数;利用 watchDog 延续锁过期时间;利用信号量控制锁重试等待

缺陷: redis宕机引起锁失效问题

🎄 Redisson 的 multiLock:

原理: 多个独立的 Redis 节点,必须在所有节点都获取重入锁,才算获取锁成功

缺陷: 运维成本高、实现复杂

相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
打赏
0
1
1
0
11
分享
相关文章
基于Scrapy-Redis的分布式景点数据爬取与热力图生成
基于Scrapy-Redis的分布式景点数据爬取与热力图生成
228 67
高并发秒杀系统实战(Redis+Lua分布式锁防超卖与库存扣减优化)
秒杀系统面临瞬时高并发、资源竞争和数据一致性挑战。传统方案如数据库锁或应用层锁存在性能瓶颈或分布式问题,而基于Redis的分布式锁与Lua脚本原子操作成为高效解决方案。通过Redis的`SETNX`实现分布式锁,结合Lua脚本完成库存扣减,确保操作原子性并大幅提升性能(QPS从120提升至8,200)。此外,分段库存策略、多级限流及服务降级机制进一步优化系统稳定性。最佳实践包括分层防控、黄金扣减法则与容灾设计,强调根据业务特性灵活组合技术手段以应对高并发场景。
377 7
【📕分布式锁通关指南 10】源码剖析redisson之MultiLock的实现
Redisson 的 MultiLock 是一种分布式锁实现,支持对多个独立的 RLock 同时加锁或解锁。它通过“整锁整放”机制确保所有锁要么全部加锁成功,要么完全回滚,避免状态不一致。适用于跨多个 Redis 实例或节点的场景,如分布式任务调度。其核心逻辑基于遍历加锁列表,失败时自动释放已获取的锁,保证原子性。解锁时亦逐一操作,降低死锁风险。MultiLock 不依赖 Lua 脚本,而是封装多锁协调,满足高一致性需求的业务场景。
77 0
【📕分布式锁通关指南 10】源码剖析redisson之MultiLock的实现
|
2月前
|
redis分布式锁在高并发场景下的方案设计与性能提升
本文探讨了Redis分布式锁在主从架构下失效的问题及其解决方案。首先通过CAP理论分析,Redis遵循AP原则,导致锁可能失效。针对此问题,提出两种解决方案:Zookeeper分布式锁(追求CP一致性)和Redlock算法(基于多个Redis实例提升可靠性)。文章还讨论了可能遇到的“坑”,如加从节点引发超卖问题、建议Redis节点数为奇数以及持久化策略对锁的影响。最后,从性能优化角度出发,介绍了减少锁粒度和分段锁的策略,并结合实际场景(如下单重复提交、支付与取消订单冲突)展示了分布式锁的应用方法。
181 3
Redis设计与实现——分布式Redis
Redis Sentinel 和 Cluster 是 Redis 高可用与分布式架构的核心组件。Sentinel 提供主从故障检测与自动切换,通过主观/客观下线判断及 Raft 算法选举领导者完成故障转移,但存在数据一致性和复杂度问题。Cluster 支持数据分片和水平扩展,基于哈希槽分配数据,具备自动故障转移和节点发现机制,适合大规模高并发场景。复制机制包括全量同步和部分同步,通过复制积压缓冲区优化同步效率,但仍面临延迟和资源消耗挑战。两者各有优劣,需根据业务需求选择合适方案。
|
2月前
|
从扣减库存场景来讲讲redis分布式锁中的那些“坑”
本文从一个简单的库存扣减场景出发,深入分析了高并发下的超卖问题,并逐步优化解决方案。首先通过本地锁解决单机并发问题,但集群环境下失效;接着引入Redis分布式锁,利用SETNX命令实现加锁,但仍存在死锁、锁过期等隐患。文章详细探讨了通过设置唯一标识、续命机制等方法完善锁的可靠性,并最终引出Redisson工具,其内置的锁续命和原子性操作极大简化了分布式锁的实现。最后,作者剖析了Redisson源码,揭示其实现原理,并预告后续关于主从架构下分布式锁的应用与性能优化内容。
117 0
|
2月前
|
Redis+Caffeine构建高性能二级缓存
大家好,我是摘星。今天为大家带来的是Redis+Caffeine构建高性能二级缓存,废话不多说直接开始~
351 0
基于Spring Data Redis与RabbitMQ实现字符串缓存和计数功能(数据同步)
总的来说,借助Spring Data Redis和RabbitMQ,我们可以轻松实现字符串缓存和计数的功能。而关键的部分不过是一些"厨房的套路",一旦你掌握了这些套路,那么你就像厨师一样可以准备出一道道饕餮美食了。通过这种方式促进数据处理效率无疑将大大提高我们的生产力。
123 32
|
2月前
|
Redis:现代服务端开发的缓存基石与电商实践-优雅草卓伊凡
Redis:现代服务端开发的缓存基石与电商实践-优雅草卓伊凡
73 5
Redis:现代服务端开发的缓存基石与电商实践-优雅草卓伊凡

热门文章

最新文章

AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等

登录插画

登录以查看您的控制台资源

管理云资源
状态一览
快捷访问