Redisson分布式锁之加解锁详解

目录
  • 引言
  • 锁的可重入性
  • 加锁
  • 锁续命
  • 释放锁

引言

2023的金三银四来的没想象中那么激烈,一个朋友前段时间投了几十家,多数石沉大海,好不容易等来面试机会,就恰好被问道项目中关于分布式锁的应用,后涉及Redisson实现分布式锁的原理,答不上来。

锁的可重入性

我们都知道,Java中synchronized和lock都支持可重入,synchronized的锁关联一个线程持有者和一个计数器。当一个线程请求成功后,JVM会记下持有锁的线程,并将计数器计为1。此时其他线程请求该锁,则必须等待。而该持有锁的线程如果再次请求这个锁,就可以再次拿到这个锁,同时计数器会递增。当线程退出一个synchronized方法/块时,计数器会递减,如果计数器为0则释放该锁;在ReentrantLock中,底层的 AQS 对应的state 同步状态值表示线程获取该锁的可重入次数,通过CAS方式进行设置,在默认情况下,state的值为0 表示当前锁没有被任何线程持有,原理类似。所以如果想要实现可重入性,可能须有一个计数器来控制重入次数,实际Redisson确实是这么做的。

好的我们通过Redisson客户端进行设置,并循环3次,模拟锁重入:000

for(int i = 0; i < 3; i++) {
    RedissonLockUtil.tryLock("distributed:lock:distribute_key", TimeUnit.SECONDS, 20, 100);
 }

连接Redis客户端进行查看:

可以看到,我们设置的分布式锁是存在一个hash结构中,value看起来是循环的次数3,key就不怎么认识了,那这个key是怎么设置进去的呢,另外为什么要设置成为Hash类型呢?

加锁

我们先来看看普通的分布式锁的上锁流程:

说明:

  • 客户端在进行加锁时,会校验如果业务上没有设置持有锁时长leaseTime,会启动看门狗来每隔10s进行续命,否则就直接以leaseTime作为持有的时长;
  • 并发场景下,如果客户端1锁还未释放,客户端2尝试获取,加锁必然失败,然后会通过发布订阅模式来订阅Key的释放通知,并继续进入后续的抢锁流程。
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
      long time = unit.toMillis(waitTime);
      long current = System.currentTimeMillis();
      long threadId = Thread.currentThread().getId();
      Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
      if (ttl == null) {
         return true;
      } else {
         // 订阅分布式Key对应的消息,监听其它锁持有者释放,锁没有释放的时候则会等待,直到锁释放的时候会执行下面的while循环
         CompletableFuture subscribeFuture = this.subscribe(threadId);
         subscribeFuture.get(time, TimeUnit.MILLISECONDS);
         try {
            do {
               // 尝试获取锁
               ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);
               // 竞争获取锁成功,退出循环,不再竞争。
               if (ttl == null) {
                  return true;
               }
               // 利用信号量机制阻塞当前线程相应时间,之后再重新获取锁
               if (ttl >= 0L && ttl < time) {
                  ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
               } else {
                  ((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture)).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
               }
               time -= System.currentTimeMillis() - currentTime;
            } while(time > 0L);
         } finally {
            // 竞争锁成功后,取消订阅该线程Id事件
            this.unsubscribe((RedissonLockEntry)this.commandExecutor.getNow(subscribeFuture), threadId);
         }
      }
   }
}
RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
        // 如果设置了持有锁的时长,直接进行尝试加锁操作
         if (leaseTime != -1L) {
            return this.tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        } else {
            // 未设置加锁时长,在加锁成功后,启动续期任务,初始默认持有锁时间是30s
            RFuture<Long> ttlRemainingFuture = this.tryLockInnerAsync(this.commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
            ttlRemainingFuture.addListener(new FutureListener<Long>() {
                public void operationComplete(Future<Long> future) throws Exception {
                    if (future.isSuccess()) {
                        Long ttlRemaining = (Long)future.getNow();
                        if (ttlRemaining == null) {
                            RedissonLock.this.scheduleExpirationRenewal(threadId);
                        }
                    }
                }
            });
            return ttlRemainingFuture;
        }
    }

我们都知道Redis执行Lua脚本具有原子性,所以在尝试加锁的下层,Redis主要执行了一段复杂的lua脚本:

-- 不存在该key时
if (redis.call('exists', KEYS[1]) == 0) then
      -- 新增该锁并且hash中该线程id对应的count置1
redis.call('hincrby', KEYS[1], ARGV[2], 1);
-- 设置过期时间
redis.call('pexpire', KEYS[1], ARGV[1]);
return nil;
end;
-- 存在该key 并且 hash中线程id的key也存在
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]);

参数说明:

KEYS[1]:对应我们设置的分布式key,即:distributed:lock:distribute_key

ARGV[1]:业务自定义的加锁时长或者默认的30s;

ARGV[2]: 具体的客户端初始化连接UUID+线程ID: 9d8f0907-1165-47d2-8983-1e130b07ad0c:1

我们从上面的脚本中可以看出核心逻辑其实不难:

  • 如果分布式锁Key未被任何端持有,直接根据“客户端连接ID+线程ID” 进行初始化设置,并设置重入次数为1,并设置Key的过期时间;
  • 否则重入次数+1,并重置过期时间;

锁续命

接下来看看scheduleExpirationRenewal续命是怎么做的呢?

private void scheduleExpirationRenewal(final long threadId) {
   if (!expirationRenewalMap.containsKey(this.getEntryName())) {
      Timeout task = this.commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
         public void run(Timeout timeout) throws Exception {
            // 执行续命操作
            RFuture<Boolean> future = RedissonLock.this.renewExpirationAsync(threadId);
            future.addListener(new FutureListener<Boolean>() {
               public void operationComplete(Future<Boolean> future) throws Exception {
                  RedissonLock.expirationRenewalMap.remove(RedissonLock.this.getEntryName());
                          ...
                  // 续命成功,继续
                  if ((Boolean)future.getNow()) {
                     RedissonLock.this.scheduleExpirationRenewal(threadId);
                  }
               }
            });
         }
      }, this.internalLockLeaseTime / 3L, TimeUnit.MILLISECONDS);
   }
}

Tip小知识点:

  • 续期是用的什么定时任务执行的?
    Redisson用netty的HashedWheelTimer做命令重试机制,原因在于一条redis命令的执行不论成功或者失败耗时都很短,而HashedWheelTimer是单线程的,系统性能开销小。

而在上面的renewExpirationAsync中续命操作的执行核心Lua脚本要做的事情也非常的简单,就是给这个Key的过期时间重新设置为指定的30s.

if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then
    redis.call('pexpire', KEYS[1], ARGV[1]);
    return 1;
end;
return 0;

释放锁

释放锁主要是除了解锁本省,另外还要考虑到如果存在续期的情况,要将续期任务删除:

public RFuture<Void> unlockAsync(long threadId) {
   // 解锁
   RFuture<Boolean> future = this.unlockInnerAsync(threadId);
   CompletionStage<Void> f = future.handle((opStatus, e) -> {
      // 解除续期
      this.cancelExpirationRenewal(threadId);
      ...
   });
   return new CompletableFutureWrapper(f);
}

在unlockInnerAsync内部,Redisson释放锁其实核心也是执行了如下一段核心Lua脚本:

    // 校验是否存在
    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
      redis.call('pexpire', KEYS[1], ARGV[2]);
      return 0;
   // 删除Key
    else redis.call('del', KEYS[1]);
      // 通知阻塞的客户端可以抢锁啦
      redis.call('publish', KEYS[2], ARGV[1]);
      return 1;
      end;
      return nil;

其中:

KEYS[1]: 分布式锁
KEYS[2]: redisson_lock_channel:{分布式锁} 发布订阅消息的管道名称
ARGV[1]: 发布的消息内容
ARGV[2]: 锁的过期时间
ARGV[3]: 线程ID标识名称

其它问题

  • 红锁这么火,但真的靠谱么?
  • Redisson公平锁是什么情况?

以上就是Redisson分布式锁第一弹-加解锁的详细内容,更多关于Redisson分布式锁加解锁的资料请关注我们其它相关文章!

(0)

相关推荐

  • Redisson如何解决Redis分布式锁提前释放问题

    目录 前言: 一.问题描述: 二.原因分析: 三.解决方案: 1.思考: 2.Redisson简单配置: 3.使用样例: 四.源码分析 1.lock加锁操作 2.unlock解锁操作 总结: 相关参考: 前言: 在分布式场景下,相信你或多或少需要使用分布式锁来访问临界资源,或者控制耗时操作的并发性. 当然,实现分布式锁的方案也比较多,比如数据库.redis.zk 等等.本文主要结合一个线上案例,讲解 redis 分布式锁的相关实现. 一.问题描述: 某天线上出现了数据重复处理问题,经排查后发现,

  • Redis中Redisson红锁(Redlock)使用原理

    目录 简介 为什么使用Redis的红锁 解决方案:使用红锁 Redisson红锁实例 Redisson红锁原理 参考文章 简介 说明 本文介绍为什么要使用Redis的红锁(Redlock).什么是Redis的红锁以及Redis红锁的原理. 本文用Redisson来介绍Redis红锁的用法. Redisson 高版本会根据redisClient的模式来决定getLock返回的锁类型,如果集群模式,满足红锁的条件,则会直接返回红锁. 官网 REDIS distlock -- Redis中国用户组(C

  • Redisson 加锁解锁的实现

    目录 分布式锁使用 getLock tryLock unLock 总结 分布式锁使用 对于 redisson 分布式锁的使用很简单: 1.调用 getLock 函数获取锁操作对象:2.调用 tryLock 函数进行加锁:3.调用 unlock 函数进行解锁: 注意 unlock 操作需要放到 finally 代码段中,保证锁可以被释放. private void sumLock() { lock = redissonClient.getLock("sum-lock"); boolean

  • Redisson如何解决redis分布式锁过期时间到了业务没执行完问题

    目录 面试问题 问题分析 如何回答 一.写在前面 二.Redisson实现Redis分布式锁的底层原理 (1)加锁机制 (2)锁互斥机制 (3)watch dog自动延期机制 (4)可重入加锁机制 (5)释放锁机制 (6)上述Redis分布式锁的缺点 总结 面试问题 Redis锁的过期时间小于业务的执行时间该如何续期? 问题分析 首先如果你之前用Redis的分布式锁的姿势正确,并且看过相应的官方文档的话,这个问题So easy.我们来看 很多同学在用分布式锁时,都是直接百度搜索找一个Redis分

  • Spring Boot 集成Redisson实现分布式锁详细案例

    目录 前言 分布式锁实现 引入jar包 Redisson的配置 application.yml中引入redisson.yml配置 redisson.yml配置 封装Redisson工具类 模拟秒杀扣减库存 测试代码 总结 前言 Spring Boot集成Redis实现单机分布式锁针对单机分布式锁还是存在锁定续期.可重入的问题,本文将采用Spring Boot 集成Ression实现分布式锁进行详细讲解. 分布式锁实现 引入jar包 <dependency> <groupId>org

  • Redisson分布式锁之加解锁详解

    目录 引言 锁的可重入性 加锁 锁续命 释放锁 引言 2023的金三银四来的没想象中那么激烈,一个朋友前段时间投了几十家,多数石沉大海,好不容易等来面试机会,就恰好被问道项目中关于分布式锁的应用,后涉及Redisson实现分布式锁的原理,答不上来. 锁的可重入性 我们都知道,Java中synchronized和lock都支持可重入,synchronized的锁关联一个线程持有者和一个计数器.当一个线程请求成功后,JVM会记下持有锁的线程,并将计数器计为1.此时其他线程请求该锁,则必须等待.而该持

  • go 分布式锁简单实现实例详解

    目录 正文 案例 资源加锁 使用redis来实现分布式锁 redis lua保证原子性 正文 其实锁这种东西,都能能不加就不加,锁会导致程序一定程度上退回到串行化,进而降低效率. 案例 首先,看一个案例,如果要实现一个计数器,并且是多个协程共同进行的,就会出现以下的情况: package main import ( "fmt" "sync" ) func main() { numberFlag := 0 wg := new(sync.WaitGroup) for i

  • redis分布式锁的实现原理详解

    首先,为了确保分布式锁可用,我们至少要确保锁的实现同时满足以下四个条件: 1.互斥性.在任意时刻,只有一个客户端能持有锁. 2.不会发生死锁.即使有一个客户端在持有锁的期间崩溃而没有主动解锁,也能保证后续其他客户端能加锁. 3.具有容错性.只要大部分的Redis节点正常运行,客户端就可以加锁和解锁. 4.解铃还须系铃人.加锁和解锁必须是同一个客户端,客户端自己不能把别人加的锁给解了. 下边是代码实现,首先我们要通过Maven引入Jedis开源组件,在pom.xml文件加入下面的代码: <depe

  • 详解Spring Cache使用Redisson分布式锁解决缓存击穿问题

    目录 1 什么是缓存击穿 2 为什么要使用分布式锁 3 什么是Redisson 4 Spring Boot集成Redisson 4.1 添加maven依赖 4.2 配置yml 4.3 配置RedissonConfig 5 使用Redisson的分布式锁解决缓存击穿 1 什么是缓存击穿 一份热点数据,它的访问量非常大.在其缓存失效的瞬间,大量请求直达存储层,导致服务崩溃. 2 为什么要使用分布式锁 在项目中,当共享资源出现竞争情况的时候,为了防止出现并发问题,我们一般会采用锁机制来控制.在单机环境

  • python多线程互斥锁与死锁问题详解

    目录 一.多线程共享全局变量 二.给线程加一把锁锁 三.死锁问题 总结 一.多线程共享全局变量 代码实现的功能: 创建work01与worker02函数,对全局变量进行加一操作创建main函数,生成两个线程,同时调用两个函数 代码如下: import threading result = 0 # 定义全局变量result def work1(num): global result for i in range(num): result += 1 print('------from work1--

  • Hadoop 分布式存储系统 HDFS的实例详解

    HDFS是Hadoop Distribute File System 的简称,也就是Hadoop的一个分布式文件系统. 一.HDFS的优缺点 1.HDFS优点: a.高容错性 .数据保存多个副本 .数据丢的失后自动恢复 b.适合批处理 .移动计算而非移动数据 .数据位置暴露给计算框架 c.适合大数据处理 .GB.TB.甚至PB级的数据处理 .百万规模以上的文件数据 .10000+的节点 d.可构建在廉价的机器上 .通过多副本存储,提高可靠性 .提供了容错和恢复机制 2.HDFS缺点 a.低延迟数

  • Spark GraphX 分布式图处理框架图算法详解

    目录 正文 Graphx图结构 1. 最短路径 示例数据 可视化数据 计算最短路径 2. 网页排名 数据可视化 pagerank算法测试 算法结果 3. 连通域(连通组件) 加载图测试连通域 生成图测试 图实例的形态展示 强连接域的计算 4. 三角计数 代码测试 测试结果 5. 标签传播算法(LPA) 基本思想 正文 Spark GraphX是一个分布式图处理框架,基于 Pregel 接口实现了常用的图算法. 包括 PageRank.SVDPlusPlus.TriangleCount. Conn

  • Redisson分布式锁的源码解读分享

    目录 前言 前置知识 分布式锁的思考 Redis订阅/发布机制 Redisson 加锁 订阅 解锁 看门狗 前言 Redisson是一个在Redis的基础上实现的Java驻内存数据网格(In-Memory Data Grid).Redisson有一样功能是可重入的分布式锁.本文来讨论一下这个功能的特点以及源码分析. 前置知识 在讲Redisson,咱们先来聊聊分布式锁的特点以及Redis的发布/订阅机制,磨刀不误砍柴工. 分布式锁的思考 首先思考下,如果我们自己去实现一个分布式锁,这个锁需要具备

  • Spring Cloud Config分布式配置中心使用介绍详解

    目录 1.分布式配置中心应用场景 2.Spring Cloud Config 2.1.Config简介 2.2.Config分布式配置应用 2.3.构建Config Server统一配置中心 2.4.构建Client客户端(在已有简历微服务基础上) 1.分布式配置中心应用场景 往往,我们使用配置文件管理⼀些配置信息,比如application.yml 单体应用架构:配置信息的管理.维护并不会显得特别麻烦,手动操作就可以,因为就一个工程: 微服务架构:因为我们的分布式集群环境中可能有很多个微服务,

  • Hadoop-3.1.2完全分布式环境搭建过程图文详解(Windows 10)

    一.前言 Hadoop原理架构本人就不在此赘述了,可以自行百度,本文仅介绍Hadoop-3.1.2完全分布式环境搭建(本人使用三个虚拟机搭建). 首先,步骤: ① 准备安装包和工具: hadoop-3.1.2.tar.gz ◦ jdk-8u221-linux-x64.tar.gz(Linux环境下的JDK) ◦ CertOS-7-x86_64-DVD-1810.iso(CentOS镜像) ◦工具:WinSCP(用于上传文件到虚拟机),SecureCRTP ortable(用于操作虚拟机,可复制粘

随机推荐