您的位置：首页 > 文旅 > 旅游 > 北京网络营销咨询公司_东莞网站建设专业品牌_站长seo_seo3的空间构型

北京网络营销咨询公司_东莞网站建设专业品牌_站长seo_seo3的空间构型

2024/12/24 10:45:14 来源：https://blog.csdn.net/qq_45956730/article/details/142874212  浏览:  次  关键词：北京网络营销咨询公司_东莞网站建设专业品牌_站长seo_seo3的空间构型
北京网络营销咨询公司_东莞网站建设专业品牌_站长seo_seo3的空间构型

目录

一、SETNX+EXPIRE

Redis分布式锁误删问题

业务阻塞造成锁超时释放问题

二、SETNX+LUA脚本

Lua脚本解决多条命令原子性问题

利用Java代码调用Lua脚本改造分布式锁

 锁无法续期问题

三、Redisson分布式锁

 背景和定义

使用方式

Redisson的大致流程

Redisson的可重入原理

 Redisson的锁重试和WatchDog机制

源码分析

整体分析

关于锁的重试机制

关于锁的续期

关于锁的释放

四、Redisson的MutiLock原理

背景和定义

MutiLock加锁原理

源码分析

整体分析

关于锁的获取

关于锁的释放

总结

分布式锁是一种用于控制在分布式系统中多个进程或线程对共享资源的访问的机制。在分布式环境中，多个进程可能需要同时访问某个共享资源，由于分布式环境中存在多个节点，传统的单机锁（如Java中的ReentrantLock）无法满足跨节点的同步需求，因此需要一种适应分布式场景的锁机制。分布式锁可以确保在任何时刻只有一个进程能够访问该资源，从而避免数据竞争和不一致性问题。

那么分布式锁应该满足一些什么条件呢？

  • 可见性：多个线程都能看到相同的结果。注意：这里说的可见性并不是并发编程中指的内存可见性，只是说多个进程之间都能感知到变化的意思。
  • 互斥：互斥是分布式锁的最基本条件，使得程序串行执行。
  • 高可用：程序不易崩溃，时时刻刻都保证较高的可用性。
  • 高性能：由于加锁本身就让性能降低，所以对于分布式锁需要他较高的加锁性能和释放锁性能。
  • 安全性：安全也是程序中必不可少的一环。

分布式锁有多种实现方式，包括基于数据库的分布式锁、基于Redis的分布式锁、基于Zookeeper的分布式锁等。本文将重点介绍如何使用Redis实现分布式锁。

常见的分布式锁有三种，

  • MySQL：本身就带有锁机制，但是由于MySQL的性能一般，所以使用MySQL作为分布式锁比较少见。
  • Redis：作为分布式锁是非常常见的一种使用方式，利用SETNX这个方法，如果插入Key成功，则表示获得到了锁，如果有人插入成功，那么其他人就回插入失败，无法获取到锁，利用这套逻辑完成互斥，从而实现分布式锁。
  • Zookeeper：也是企业级开发中较好的一种实现分布式锁的方案。

一、SETNX+EXPIRE

实现分布式锁时需要实现两个基本方法：

  1. 获取锁
    • 互斥：确保只能有一个线程获取锁
    • 非阻塞：尝试一次，成功返回true，失败返回false
  2. 释放锁
    • 手动释放
    • 超时释放：获取锁的时候添加一个超时时间

核心思路：我们利用redis的SETNX方法，当有多个线程进入时，我们就利用该方法来获取锁。第一个线程进入时，redis 中就有这个key了则返回1。如果返回结果是1则表示抢到了锁去执行业务，然后再删除锁，退出锁逻辑。如果返回结果是0则没有抢到锁，等待一定时间之后重试。

Redis分布式锁误删问题

Redis分布式锁误删情况的逻辑说明：

  • 持有锁的线程1在锁的内部出现了阻塞，导致他的锁TTL到期，自动释放
  • 此时线程2也来尝试获取锁，由于线程1已经释放了锁，所以线程2可以拿到
  • 但是现在线程1阻塞完了，继续往下执行，要开始释放锁了
  • 那么此时就会将属于线程2的锁释放，这就是误删别人锁的情况

解决方案就是在每个线程存入锁的时候放入自己的线程标识，在删除锁的时候，判断当前这把锁是不是自己存入的，如果是则进行删除，如果不是则不进行删除。

假设还是上面的情况，线程1阻塞，锁自动释放，线程2进入到锁的内部执行逻辑，此时线程1阻塞完了，继续往下执行，开始删除锁，但是线程1发现这把锁不是自己的，所以不进行删除锁的逻辑，当线程2执行到删除锁的逻辑时，如果TTL还未到期，则判断当前这把锁是自己的，于是删除这把锁。

业务阻塞造成锁超时释放问题

更为极端的误删逻辑说明：

  • 假设线程1已经获取了锁，在判断标识一致之后，准备释放锁的时候，又出现了阻塞（例如JVM垃圾回收机制），于是锁的TTL到期了，自动释放了。
  • 那么现在线程2趁虚而入，拿到了一把锁。
  • 但是线程1的逻辑还没执行完，那么线程1就会执行删除锁的逻辑，但是在阻塞前线程1已经判断了标识一致，所以现在线程1把线程2的锁给删了，那么就相当于判断标识那行代码没有起到作用。

这就是删锁时的原子性问题，因为线程1的拿锁，判断标识和删锁不是原子操作，所以我们要防止刚刚的情况。

二、SETNX+LUA脚本

Lua脚本解决多条命令原子性问题

Redis提供了Lua脚本功能，在一个脚本中编写多条Redis命令，确保多条命令执行时的原子性。Lua是一种编程语言，可以使用Lua去操作Redis，而且还能保证它的原子性，这样拿锁，判断标识，删锁就可以实现是一个原子性动作了。

1）Redis提供的调用函数语法如下，

redis.call('命令名称','key','其他参数', ...)## 先执行set name David
redis.call('set', 'name', 'David')
## 再执行get name
local name = redis.call('get', 'name')
## 返回
return name

2）写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下，3）如果脚本中的key和value不想写死，可以作为参数传递，key类型参数会放入KEYS数组，其他参数会放入ARGV数组，在脚本中可以从KEYS和ARGV数组中获取这些参数。

原逻辑：

@Override
public void unlock() {// 获取当前线程的标识String threadId = ID_PREFIX + Thread.currentThread().getId();// 获取锁中的标识String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);// 判断标识是否一致if (threadId.equals(id)) {// 释放锁stringRedisTemplate.delete(KEY_PREFIX + name);}
}

改为，

-- 线程标识
local threadId = "UUID-31"
-- 锁的key
local key = "lock:order:userId"
-- 获取锁中线程标识
local id = redis.call('get', key)
-- 比较线程标识与锁的标识是否一致
if (threadId == id) then-- 一致则释放锁 del keyreturn redis.call('del', key)
end
return 0-- 这里的KEYS[1]就是传入锁的key
-- 这里的ARGV[1]就是线程标识
-- 比较锁中的线程标识与线程标识是否一致
if (redis.call('get', KEYS[1]) == ARGV[1]) then-- 一致则释放锁return redis.call('del', KEYS[1])
end
return 0

利用Java代码调用Lua脚本改造分布式锁

在RedisTemplate中，可以利用execute方法去执行lua脚本，

 锁无法续期问题

在使用SETNX实现的分布式锁中，存在锁无法续期导致并发冲突的问题。

假设线程A获取了锁，但是由于网络原因，执行时间超过了设置的过期时间，这是锁被释放了，线程B获取锁成功，此时线程A和B都会执行临界区的代码，这是绝对不允许的。

三、Redisson分布式锁

在使用SETNX实现的分布式锁中，存在锁无法续期导致并发冲突的问题。不过这个问题在Redisson中用看门狗的机制巧妙地解决了，这也是我们实现分布式锁最常用的方式。

 背景和定义

基于SETNX实现的分布式锁存在以下问题：

（1）不可重入问题

重入问题是指获取锁的线程，可以再次进入到相同的锁的代码块中，可重入锁的意义在于防止死锁，例如在HashTable这样的代码中，它的方法都是使用synchronized修饰的，加入它在一个方法内调用另一个方法，如果此时是不可重入的，那就死锁了。所以可重入锁的主要意义是防止死锁，我们的synchronized和Lock锁都是可重入的。

（2）不可重试

我们编写的分布式锁只能尝试一次，失败了就返回false，没有重试机制。但合理的情况应该是：当线程获取锁失败后，他应该能再次尝试获取锁。

（3）超时释放

我们在加锁的时候增加了TTL，这样我们可以防止死锁，但是如果卡顿(阻塞)时间太长，也会导致锁的释放。虽然我们采用Lua脚本来防止删锁的时候，误删别人的锁，但现在的新问题是没锁住，也有安全隐患。

（4）主从一致性

如果Redis提供了主从集群，那么当我们向集群写数据时，主机需要异步的将数据同步给从机，万一在同步之前，主机宕机了(主从同步存在延迟，虽然时间很短，但还是发生了)，那么又会出现死锁问题。

那么什么是Redisson呢？

Redisson是一个在Redis的基础上实现的Java驻内存数据网格(In-Memory Data Grid)。它不仅提供了一系列的分布式Java常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

Redis提供了分布式锁的多种多样功能，

  • 可重入锁(Reentrant Lock)
  • 公平锁(Fair Lock)
  • 联锁(MultiLock)
  • 红锁(RedLock)
  • 读写锁(ReadWriteLock)
  • 信号量(Semaphore)
  • 可过期性信号量(PermitExpirableSemaphore)
  • 闭锁(CountDownLatch)

使用方式

导入依赖，

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

配置Redisson客户端，在config包下新建RedissonConfig类，

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;@Configuration
public class RedissonConfig {@Beanpublic RedissonClient redissonClient() {Config config = new Config();config.useSingleServer().setAddress("redis://101.XXX.XXX.160:6379").setPassword("root");return Redisson.create(config);}
}

使用Redisson的分布式锁，

@Resource
private RedissonClient redissonClient;@Test
void testRedisson() throws InterruptedException {//获取可重入锁RLock lock = redissonClient.getLock("anyLock");//尝试获取锁，三个参数分别是：获取锁的最大等待时间(期间会重试)，锁的自动释放时间，时间单位boolean success = lock.tryLock(1,10, TimeUnit.SECONDS);//判断获取锁成功if (success) {try {System.out.println("执行业务");} finally {//释放锁lock.unlock();}}
}

Redisson的大致流程

在深入代码前，我们先看下加锁、看门狗续期大致的流程，有个大致印象。

Redisson的可重入原理

在分布式环境中，可重入锁（Reentrant Lock）是一个重要的概念。它指的是同一个线程可以多次获取同一把锁，而不会导致死锁。这种特性在并发编程中非常有用，因为它允许线程在执行过程中多次调用需要加锁的代码段，而不需要在每次调用时都释放和重新获取锁。

Redisson的可重入锁实现基于Redis的Lua脚本和数据结构，通过条件判断和过期时间机制实现了锁的获取和释放。同时，通过允许同一个线程多次获取同一把锁，实现了可重入性。这种实现方式既简单又高效，适用于分布式环境中的并发编程场景。

下面我们将详细分析Redisson实现可重入锁的原理：

（1）锁的获取

当线程尝试获取锁时，Redisson会向Redis发送一个Lua脚本，该脚本会在Redis中创建一个锁对象（通常是一个键值对），并设置锁的过期时间。如果锁对象不存在（即锁未被其他线程持有），则当前线程获取锁成功。如果锁对象已存在，则当前线程会尝试获取锁对象的所有权，如果成功，则获取锁成功；否则，当前线程将等待锁释放。

在Lua脚本中，Redisson使用了Redis的SETNX命令来尝试设置锁对象。SETNX命令只有在键不存在时才设置该键的值，因此它可以用来实现锁的获取。同时，Redisson还使用了Redis的EXPIRE命令来设置锁的过期时间，以防止因线程崩溃或其他原因导致锁无法释放。

（2）锁的释放

当线程释放锁时，Redisson同样会向Redis发送一个Lua脚本。该脚本会检查当前线程是否持有锁对象的所有权，如果是，则删除锁对象，释放锁。如果不是，则不做任何操作。

在Lua脚本中，Redisson使用了Redis的GET命令来获取锁对象的值，并检查该值是否与当前线程的标识相匹配。如果相匹配，则说明当前线程持有锁对象的所有权，可以使用DEL命令删除锁对象来释放锁。否则，说明当前线程不持有锁对象的所有权，无法释放锁。

（3）可重入性的实现

Redisson的可重入锁实现的关键在于当线程已经持有锁对象的所有权时，再次获取锁时不会进行任何操作。这是通过在Lua脚本中添加条件判断来实现的。当线程尝试获取锁时，Lua脚本会先检查锁对象是否已存在且其值是否与当前线程的标识相匹配。如果是，则说明当前线程已经持有锁对象的所有权，无需再次获取锁；否则，按照上述流程尝试获取锁。

通过这种方式，Redisson实现了可重入锁的功能。同一个线程可以多次获取同一把锁，而不会导致死锁。同时，由于使用了Redis的Lua脚本和过期时间机制，Redisson的可重入锁还具有高可用性和可配置性。

在分布式锁中，它采用hash结构来存储锁，其中外层key表示这把锁是否存在，内层key则记录当前这把锁被哪个线程持有。

 Redisson的锁重试和WatchDog机制

源码分析

整体分析

首先，会进入RedissionLock类的tryLock函数，

其中，waitTime是获取锁失败之后的等待时间，leaseTime是获取锁成功之后，对锁的剩余持有时间，默认值-1表示30s。

进一步进入tryLock函数内部，

这段 Java 代码实现了一个带有超时时间和锁持有时间限制的 tryLock 方法。该方法尝试获取一个分布式锁，并在指定时间内等待锁的获取。以下是代码的主要逻辑：

  1. 初始化时间变量：
    1. 将传入的 waitTime 转换为毫秒单位。
    2. 获取当前时间戳 current 和当前线程 ID。
  2. 尝试立即获取锁：
    1. 调用 tryAcquire 方法尝试获取锁。
    2. 如果成功 (ttl == null)，则直接返回 true 表示获取锁成功。
  3. 计算剩余等待时间：
    1. 更新 time 变量以减去已经消耗的时间。
    2. 如果 time 已经小于等于 0，则调用 acquireFailed 方法并返回 false。
  4. 订阅锁变化通知：
    1. 创建一个 RFuture 对象来订阅锁的状态变化。
    2. 等待订阅操作完成或超时。
    3. 如果订阅超时，则取消订阅操作，并调用 acquireFailed 方法返回 false。
  5. 循环尝试获取锁：
    1. 在循环中不断尝试获取锁。
    2. 每次尝试后更新剩余等待时间 time。
    3. 如果成功获取锁，则返回 true。
    4. 如果剩余时间不足，则调用 acquireFailed 方法返回 false。
    5. 否则，根据 ttl 的值决定等待时间，并尝试获取锁的通知信号。
  6. 清理资源：
    1. 最终确保取消订阅操作，释放资源。
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();//尝试获取锁（与上方）拿到null标识获取锁成功，拿到有效期标识获取锁失败Long ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);if (ttl == null) {//拿到null标识获取锁成功return true;} else {//拿到有效期标识获取锁失败//得到剩余等待时间time -= System.currentTimeMillis() - current;if (time <= 0L) {//没有剩余时间，如果执行获取锁的操作太久，锁剩余时间小于0，获取锁失败this.acquireFailed(waitTime, unit, threadId);return false;} else {//有剩余时间current = System.currentTimeMillis();//获取锁失败再次获取大概率还是失败，因此采用订阅的机制等待锁发送通知//订阅别人释放锁的信号，释放锁的lua脚本中有publish命令就是在发送消息通知RFuture<RedissonLockEntry> subscribeFuture = this.subscribe(threadId);//尝试等待结果，等到锁的剩余时间if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {//等待超过了这个时间，首先取消订阅，并且获取锁失败if (!subscribeFuture.cancel(false)) {subscribeFuture.onComplete((res, e) -> {if (e == null) {//取消订阅this.unsubscribe(subscribeFuture, threadId);}});}this.acquireFailed(waitTime, unit, threadId);//获取锁失败return false;} else {try {//又获取剩余时间，主要是等待会消耗时间，所以这里又计算了一次time -= System.currentTimeMillis() - current;if (time <= 0L) {//没有剩余时间，返回获取锁失败this.acquireFailed(waitTime, unit, threadId);boolean var20 = false;return var20;} else {//有剩余时间，可以重试boolean var16;do {long currentTime = System.currentTimeMillis();ttl = this.tryAcquire(waitTime, leaseTime, unit, threadId);if (ttl == null) {//等于null成功var16 = true;return var16;}//再看一下剩余时间还剩下多少time -= System.currentTimeMillis() - currentTime;if (time <= 0L) {this.acquireFailed(waitTime, unit, threadId);//没有剩余时间返回失败var16 = false;return var16;}//如果还有再试一次currentTime = System.currentTimeMillis();//有两种情况if (ttl >= 0L && ttl < time) {//如果ttl小于time就等ttl时间((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {//如果ttl大于time就等time时间((RedissonLockEntry)subscribeFuture.getNow()).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}time -= System.currentTimeMillis() - currentTime;} while(time > 0L);//如果时间还有那就再试一次//如果时间没有了就返回falsethis.acquireFailed(waitTime, unit, threadId);var16 = false;return var16;}} finally {//关闭订阅this.unsubscribe(subscribeFuture, threadId);}}}}
}

关于锁的重试机制

抢锁过程中，获得当前线程，通过tryAcquire进行抢锁。如果加锁失败，则while循环不断尝试。

关于锁的续期

假如目前获取锁成功了，并且锁有一个剩余的有效期，万一业务阻塞了，TTL到期了，其他线程又进来拿锁，导致线程安全问题。我必须保证我的锁是执行完业务释放的，而不是业务阻塞导致锁过期释放的，这个问题如何解决呢？

答案是有一种不断刷新任务有效期的看门狗机制，每过10秒钟自动刷新任务有效期，从而避免了业务因为线程阻塞导致锁过期自动释放，导致其他线程趁虚而入导致的线程安全问题。

我们可以继续深入源码了解，首先进入上图中的tryAcquire函数中，发现调用了tryAcquireAsync函数，又进入到tryAcquireAsync函数内部。

tryAcquireAsync 解决了ttl过期造成线程安全问题。在持有锁的剩余过期时间快使用完，而任务还未执行完成就会刷新锁的过期时间，启动续锁任务，定期刷新锁的过期时间。

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {// 如果存活时间不等于-1，表示设置了时间if (leaseTime != -1) {return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}// 如果存活时间等于-1，表示未设置时间，使用锁看门狗超时时间作为默认存活时间RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime, commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);// 添加监听器，用于处理锁的获取结果    ttlRemainingFuture.onComplete((ttlRemaining, e) -> {if (e != null) { // return;}// 如果成功获取锁（ttlRemaining为null表示成功）if (ttlRemaining == null) {// 启动续锁任务，定期刷新锁的过期时间，scheduleExpirationRenewal(threadId);}});// 返回包含锁的剩余超时时间的Futurereturn ttlRemainingFuture;
}

详细分析如下。

（1）代码中会有一个条件分支，因为lock方法有重载方法，一个是带参数，一个是不带参数。所以如果传入了参数，此时leaseTime != -1 则会进去抢锁。

如果是没有传入时间，则此时也会进行抢锁， 抢锁时间是默认看门狗过期时间 commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()。

其中的tryLockInnerAsync方法是选择slot槽并执行lua脚本。返回null说明加锁成功，反之失败。如果设置了过期时间，第二个参数就传设置的时间。反之，使用默认过期时间internallockLeaseTime=30s。

  • KEYS[1]：表示redis中锁的名称
  • ARGV[2]：表示进程的唯一标识，客户端UUID与线程id的组合

LUA脚本的流程是首先判断锁的key是否存在，以及如果锁的key不存在则判断是否是同一线程再次过来对同一个key进行加锁，也就是当前key是否被当前线程持有（可重入性）。如果上述两个条件任意一个成立，则对当前key执行自增和设置过期时间操作，并返回null表示加锁成功。反之返回当前锁的过期时间，表示加锁失败。

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {internalLockLeaseTime = unit.toMillis(leaseTime);return evalWriteAsync(getName(), LongCodec.INSTANCE, command,// 检查锁的键KEYS[1]是否存在。如果不存在则表示锁没有被占用"if (redis.call('exists', KEYS[1]) == 0) then " +// 将锁计数器加 1"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +// 设置锁的过期时间"redis.call('pexpire', KEYS[1], ARGV[1]); " +// 返回 nil，表示锁被成功获取"return nil; " +"end; " +// 通过ARGV[2]表示的线程ID,检查特定线程是否已经持有该锁"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +// 将锁计数器加 1"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +// 设置锁的过期时间"redis.call('pexpire', KEYS[1], ARGV[1]); " +// 返回 nil，表示锁被成功获取"return nil; " +"end; " +// 如果锁已被其他线程占用（即 exists 返回 1），则调用 pttl 命令返回当前锁的剩余生存时间（TTL）"return redis.call('pttl', KEYS[1]);",Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}

（2）

接下来，ttlRemainingFuture.onComplete((ttlRemaining, e) 这句话相当于对以上抢锁进行了监听，也就是说当逻辑执行完成后，此方法会被调用。具体调用逻辑就是如果ttlRemaining为null，表示成功获取锁，就会去后台开启一个看门狗线程，进行续约逻辑。

（3）其中，进入scheduleExpirationRenewal方法，

private void scheduleExpirationRenewal(long threadId) {// 创建一个新的ExpirationEntry对象，用于跟踪当前线程的过期续租信息ExpirationEntry entry = new ExpirationEntry();// 将该对象作为值放入EXPIRATION_RENEWAL_MAP中时，如果已经存在与getEntryName()返回的键对应的值，则不会覆盖现有值，并将其返回给 oldEntryExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);// 该键在映射中已经存在，则将当前线程ID添加到现有条目中if (oldEntry != null) {oldEntry.addThreadId(threadId); // 该键在映射中不存在，当前线程 ID 添加到新条目中，并续租锁的过期时间} else { entry.addThreadId(threadId);renewExpiration();}}

首先创建一个新的ExpirationEntry对象，将该对象作为值放入EXPIRATION_RENEWAL_MAP中，EXPIRATION_RENEWAL_MAP存放续期任务（entryName—>K,entry----V）。

  • 键为特定的getEntryName()返回值。entryName是连接名和锁名组合成的。 不同的业务组成了不同的锁名称，因此可以相当于不同业务的锁是独立的，在MAP中互不干扰。
  • 值为ExpirationEntry对象。

解释一下EXPIRATION_RENEWAL_MAP.putIfAbsent()，如果MAP里面存在该key则返回key对应的Value，如果不存在才直接插入kv。不会出现后来的val覆盖先前的val的情况。如果使用的是MAP.put()，在key重复插入的情况下会导致后来的覆盖前面的val的情况。这在可重入锁中是不能发生的。

因此，putIfAbsent帮助实现了可重入锁的功能：不管后来的锁重入几次，保证获得的都是之前的entry。并且如果这个key是新插入的话，会额外执行一个更新有效期的操作renewExpiration()。

（4）那继续进入renewExpiration()方法，它主要实现了一个锁的续期机制，使用netty的时间轮进行续期，确保在锁的有效期内能够延续锁的生存时间。

private void renewExpiration() {// 1、首先会从EXPIRATION_RENEWAL_MAP中获取一个值，如果为null说明锁可能已经被释放或过期，因此不需要进行续期，直接返回ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee == null) {return;}// 2、基于TimerTask实现一个定时任务，设置internalLockLeaseTime / 3的时长进行一次锁续期，也就是每10s进行一次续期。Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {// 从EXPIRATION_RENEWAL_MAP里获取一个值，检查锁是否被释放ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());// 如果为null则说明锁也被释放了，不需要续期if (ent == null) {return;}// 如果不为null，则获取第一个thread（也就是持有锁的线程）Long threadId = ent.getFirstThreadId();if (threadId == null) {return;}// 如果threadId 不为null，说明需要续期，它会异步调用renewExpirationAsync(threadId)方法来实现续期RFuture<Boolean> future = renewExpirationAsync(threadId);// 处理结果future.onComplete((res, e) -> {// 如果有异常if (e != null) {log.error("Can't update lock " + getName() + " expiration", e);return;}// 如果续期成功，则会重新调用renewExpiration()方法进行下一次续期if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}

首先，从 EXPIRATION_RENEWAL_MAP 中获取当前锁的 ExpirationEntry 对象。如果该对象为null，说明锁可能已经被释放或过期，因此不需要进行续期，直接返回。 如果不是null，创建一个定时任务。定时任务使用 TimerTask 来实现，定时续期的时长设置为 internalLockLeaseTime / 3，如果internalLockLeaseTime 为 30 秒，意味着每 10 秒进行一次续期。

在定时任务的 run() 方法中，再次从 EXPIRATION_RENEWAL_MAP 中获取锁的状态。如果此时的 ent 为 null，表示锁已经被释放，任务不需要再继续续期，直接返回。

然后，调用 getFirstThreadId() 方法获取持有锁的第一个线程的 ID。如果 threadId 为 null，同样表示没有持锁线程，不需要续期，直接返回。 如果 threadId 不为 null，则调用 renewExpirationAsync(threadId) 异步续期锁，返回一个 RFuture<Boolean> 对象，表示续期操作的结果。 使用 onComplete 方法处理续期操作的结果，如果res=true表示续期成功，则再次调用 renewExpiration() 方法以安排下一次续期。

（5）renewExpiration()函数内部的RFuture<Boolean> future = renewExpirationAsync(threadId);又是一个关键的函数，跳入renewExpirationAsync(threadId)内部一探究竟。

这个renewExpirationAsync()是一个异步刷新有效期的函数，它主要是用evaLWriteAsync方法来异步执行一段Lua脚本，重置当前threadId线程持有的锁的有效期。

  • ARGV[1]：引用传入的第一个非键参数，表示希望设置的新过期时间（毫秒），锁的默认租约时间为internalLockLeaseTime。
  • ARGV[2]：引用传入的第二个非键参数，表示通过getLockName(threadId)根据线程ID生成特定的锁标识符，确保操作的是特定线程的锁。

Lua脚本中，如果当前key存在，说明当前锁还被该线程持有，那么就重置过期时间为30s，并返回true表示续期成功，反之返回false。

需要注意的是：这里更新完有效期之后又递归调用自己，10秒钟之后又刷新有效期。

此逻辑就是续约逻辑，最后这个任务的执行流程大概是这样的。因为锁的失效时间是30s，当10s之后，此时这个 timeTask 就触发了，就去进行续约，把当前这把锁续约成30s。如果操作成功，那么此时就会递归调用自己，再重新设置一个timeTask()，于是再过10s后又再设置一个timerTask，完成不停的续约。 重复循环往返！ 

（6）理解到这层意图，我们又跳回到前面的scheduleExpirationRenewal方法，

关于锁的释放

那问题来了，这个任务什么时候才会释放呢？什么时候才会取消呢？这当然是在释放锁的时候，这个任务才会被释放！

（1）进入RedissionLock类的unlock方法内，unlock方法用于释放锁。

继续进入unlockAsync方法内部，这个方法用于异步解锁，提供了一种安全且异步的方式来尝试释放一个分布式锁，同时妥善处理了各种可能的错误情况，确保了锁管理的健壮性。

（3）其中，cancelExpirationRenewal方法，主要功能是取消与特定线程相关的过期续期任务。

实现了灵活的过期续期任务管理，既支持按需移除特定线程的关注，也能够彻底取消整个续期任务，确保系统资源得到有效管理和释放。

最后总结一下，如何解决的业务因为线程阻塞导致业务还没执行完就锁过期自动释放，导致其他线程趁虚而入导致的线程安全问题？通过这一段上面的源码剖析，我们知道是有一种不断刷新任务有效期的看门狗机制，没过10秒钟自动刷新任务有效期，从而避免了业务因为线程阻塞导致锁过期自动释放，导致其他线程趁虚而入导致的线程安全问题。

我们在最后，总结分布式锁的流程图加深印象和理解。

四、Redisson的MutiLock原理

背景和定义

为了提高Redis的可用性，我们会搭建集群或者主从，现在以主从为例。

此时我们去写命令，写在主机上，主机会将数据同步给从机，但是假设主机还没来得及把数据写入到从机去的时候，主机宕机了。哨兵会发现主机宕机了，于是选举一个slave(从机)变成master(主机)，而此时新的master(主机)上并没有锁的信息，那么其他线程就可以获取锁，又会引发安全问题。

为了解决这个问题。Redisson提出来了MutiLock锁。Rredisson解决得比较粗暴，直接取消了主从关系，所有的节点都是独立关系，如果一台redis服务器宕机了也需要从其他redis服务器获取锁，全部获取成功才成功。

既然主从会存在数据一致性问题，那么Redisson就放弃了主从。每个节点的地位都是一样的，都可以当做是主机。当一个获取锁请求过来后，Redisson就将这个加锁请求，发给集群中的所有Redisson机器，集群中的机器获取请求后，就开始执行加锁操作。当所有机器返回加锁成功，Redisson才认为这个锁是成功的。反之则获取锁失败。

为了保证可用性，可以给这些机器都加从机。如果加了从机器，那么就会存在主从同步问题，会不会锁失效呢？是不会的。

假设第一台的主挂了，对应的slave成了master。此时，新的master是没有锁的，有个加锁的请求过来，发到三个机器后，其中第一台新master加锁成功了，但是剩下两台，加锁失败了。这种情况下，Redisson就会任务加锁失败。

这就是Redisson提出的MutiLock锁思想。当使用这把锁的时候，就不使用主从了，每个节点的位置都是一样的，这把锁加锁的逻辑需要写入到每一个主从节点上，只有所有的服务器都写入成功，此时才是加锁成功的。

其实就是将多个锁合并成了一个大锁，对一个大锁进行统一的管理。

MutiLock加锁原理

我们可以先使用虚拟机额外搭建两个Redis节点，注入三个RedissonClient对象，然后创建联锁。

当我们去设置了多个锁时，redission会将多个锁添加到一个集合中，然后用while循环去不停去尝试拿锁，但是会有一个总共的加锁时间，这个时间是用需要加锁的个数 * 1500ms ，假设有3个锁，那么时间就是4500ms，假设在这4500ms内，所有的锁都加锁成功， 那么此时才算是加锁成功，如果在4500ms有线程加锁失败，则会再次去进行重试。

源码分析

整体分析

在源码中，当我们没有传入锁对象来创建联锁的时候，会抛出一个异常，反之则将我们传入的可变参数锁对象封装成一个集合，

public RedissonMultiLock(RLock... locks) {if (locks.length == 0) {throw new IllegalArgumentException("Lock objects are not defined");} else {this.locks.addAll(Arrays.asList(locks));}
}

然后，进入RedissonMutiLock类的tryLock()方法，在RedissonMultiLock对象中tryLock代码如下，

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {long newLeaseTime = -1L;//如果传入了释放时间if (leaseTime != -1L) {//再判断一下是否有等待时间if (waitTime == -1L) {  //如果没传等待时间，不重试，则只获得一次newLeaseTime = unit.toMillis(leaseTime);} else { //想要重试，耗时较久，万一释放时间小于等待时间，则会有问题，所以这里将等待时间乘以二newLeaseTime = unit.toMillis(waitTime) * 2L;}}//获取当前时间long time = System.currentTimeMillis();//剩余等待时间long remainTime = -1L;if (waitTime != -1L) remainTime = unit.toMillis(waitTime);//计算锁等待时间，与剩余等待时间一样    long lockWaitTime = this.calcLockWaitTime(remainTime);//锁失败的限制，源码返回是的0int failedLocksLimit = this.failedLocksLimit();//已经获取成功的锁List<RLock> acquiredLocks = new ArrayList(this.locks.size());//迭代器，用于遍历所有独立的锁ListIterator<RLock> iterator = this.locks.listIterator();while(iterator.hasNext()) {//拿到每一个锁RLock lock = (RLock)iterator.next();boolean lockAcquired;try {//没有等待时间和释放时间，调用空参的tryLockif (waitTime == -1L && leaseTime == -1L)  lockAcquired = lock.tryLock();else { //否则调用带参的tryLocklong awaitTime = Math.min(lockWaitTime, remainTime);lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);}} catch (RedisResponseTimeoutException var21) {this.unlockInner(Arrays.asList(lock));lockAcquired = false;} catch (Exception var22) {lockAcquired = false;}//判断获取锁是否成功，成功则将锁放入成功锁的集合if (lockAcquired)  acquiredLocks.add(lock);else {//如果获取锁失败//判断当前锁的数量减去已经成功获取锁的数量，如果为0则所有锁都成功获取，跳出循环if (this.locks.size() - acquiredLocks.size() == this.failedLocksLimit())  break;//否则将拿到的锁都释放掉if (failedLocksLimit == 0) {this.unlockInner(acquiredLocks);//如果等待时间为-1，则不想重试，直接返回falseif (waitTime == -1L)  return false;failedLocksLimit = this.failedLocksLimit();//将已经拿到的锁都清空acquiredLocks.clear();//将迭代器往前迭代，相当于重置指针，放到第一个然后重试获取锁while(iterator.hasPrevious())  iterator.previous();} else {--failedLocksLimit;}}//如果剩余时间不为-1，很充足if (remainTime != -1L) {//计算现在剩余时间remainTime -= System.currentTimeMillis() - time;time = System.currentTimeMillis();//如果现在剩余时间为负数，则获取锁超时了if (remainTime <= 0L) {//将之前已经获取到的锁释放掉，并返回falsethis.unlockInner(acquiredLocks);//联锁成功的条件是：每一把锁都必须成功获取，一把锁失败，则都失败return false;}}}//现在所有的锁都已经拿到//如果设置了锁的有效期if (leaseTime != -1L) {List<RFuture<Boolean>> futures = new ArrayList(acquiredLocks.size());//遍历已经获取成功的锁Iterator var24 = acquiredLocks.iterator();while(var24.hasNext()) {RLock rLock = (RLock)var24.next();//重新设置每一把锁的有效期RFuture<Boolean> future = ((RedissonLock)rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);futures.add(future);}var24 = futures.iterator();while(var24.hasNext()) {RFuture<Boolean> rFuture = (RFuture)var24.next();rFuture.syncUninterruptibly();}}//但如果没设置有效期，则会触发WatchDog机制，自动帮我们设置有效期，所以大多数情况下，我们不需要自己设置有效期return true;
}

根据理解，画图如下：

总体而言，就是将 key1、key2、key3 …… keyN 放到一个 List 集合中，然后迭代循环加锁，直到所有的都成功。

关于锁的获取

我们重点分析一下tryLock()方法，来一探究竟。

（1）

首先，根据leaseTime和waitTime来设置新的过期时间newleaseTime。如果leaseTime不是-1，如果waitTime不是-1说明设置了等待时间和过期时间，则newLeaseTime=leaseTime。如果waitTime是-1说明只设置了过期时间，由于从申请获取锁到真正获取到锁是有时间消耗的，为了防止获取到的锁不至于立马过期，所以newleaseTime=leaseTime*2，因为从代码最后释放锁的逻辑来看，这里的过期时间多长，并不会影响最后锁的统一释放。

然后，计算总的剩余等待时间remainTime。如果waittime不是-1说明设置了等待时间，则remainTime=waitTime。

（2）接下来，调用calcLockWaitTime(remainTime) 计算每个锁的实际等待时间 lockWaitTime。可以看到，在联锁的场景下，锁的等待时间 lockWaitTime就等于剩余等待时间remainTime。

调用failedLocksLimit()来计算允许获取锁失败的次数failedLocksLimit=0。可以看到，在联锁的场景下，固定为0。

（3）然后，开始循环每个redis客户端，去获取锁。

创建一个acquiredLocks 集合用来报错已经获取成功的锁。

接下来，开始for循环遍历locks，locks是所有锁的集合，应该依次获取每一把锁。

当拿到每一个锁时，如果等待时间和有效时间都没有设置，就意味着只尝试一次获取锁不重试，就使用默认的方式去获取锁，这儿可以使用RedissonLock.tryLock或者其他实现了RLock接口的类对应的tryLock，简单调用tryLock()尝试获取锁。否则，传了waitTime就意味着有重试次数，就取lockWaitTime和remainTime的最小值作为可等待的时间 awaitTime，调用lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS)方法。最终，会获得获取锁的结果lockAcquired，true表示获取锁成功，反之false表示获取锁失败。

如果，任意一台redis服务器响应超时了，就应该释放所有锁，则调用unlockInner(Arrays.asList(lock))方法，用于解锁一组分布式锁。

/*** 内部解锁方法，用于批量异步解锁* 该方法通过异步方式解锁每个锁，并确保所有锁都已解锁完成* * @param locks 需要解锁的锁集合*/
protected void unlockInner(Collection<RLock> locks) {// 存储每个锁的解锁未来对象List<RFuture<Void>> futures = new ArrayList<>(locks.size());// 遍历锁集合，异步解锁每个锁，并将解锁未来对象存储到列表中for (RLock lock : locks) {futures.add(lock.unlockAsync());}// 遍历解锁未来对象列表，确保每个锁的解锁操作都已完成for (RFuture<Void> unlockFuture : futures) {unlockFuture.awaitUninterruptibly();}
}

目前， 代码进行到这儿了，

拿到结果之后，往下走，会判断lockAcquired，判断锁是否获取成功，如果加锁成功，则将成功的锁放进 acquiredLocks 集合中。

如果加锁失败，判断锁的总数locks.size()减去已经获取锁的数量acquiredLocks.size()是否等于锁失败次数的上限 failedLocksLimit()=0。因为这里failedLocksLimit是 0，只有已经获取锁的数量等于锁的总数时，这个条件表达式成立，才能跳出循环。换言之，只有把所有的锁都拿到了，才能结束for循环。往下走，需要判断 failedLocksLimit，因为这里failedLocksLimit是 0，所以会直接对成功加锁集合 acquiredLocks 中的所有锁执行锁释放。然后判断waitTime如果是-1说明不想重试，直接返回false结束。如果waitTime不是-1说明想重试，先把已经获取到的锁全部清空，并且重置迭代器。

现在，代码走到if (remainTime != -1)代码中去，判断剩余等待时间remainTime是否等于-1。如果不等于-1说明现在剩余时间很充足就可以继续拿下一把锁，并且用当前时间减去开始时间来计算获取锁的耗时时间，用剩余等待时间remainTime减去获取锁的耗时时间，就可以更新剩余等待时间remainTime。如果剩余等待时间remainTime小于等于0，证明刚才获取锁已经把剩余等待时间耗尽了，代表获取锁超时了，返回false获取锁失败。

注意，在返回false之前，需要调用unlockInner(acquiredLocks)把acquiredLocks集合中的已经获取成功的锁释放掉。

到这儿，这次for循环就结束了，进行下一次for循环。直到把所有的锁都拿到为止，这样acquiredLocks集合中就拿到了所有锁。

（4）走到这里说明全部锁都获取成功了。

如果过期时间leaseTime等于-1意味着没有自己指定了锁的过期时间，那上面获取锁的时候会走看门狗机制一直去续期。如果过期时间leaseTime不等于-1意味着自己指定了锁的过期时间，那么就需要执行expireAsync方法给每把锁都重置有效期。

为什么要执行expireAsync方法给每把锁都重置有效期呢？这是因为我们在上面获取锁的过程中，会有多个reids需要依次获取，第一把获取的锁在获取之后立即开始倒计时了，而最后一把获取的锁是刚开始倒计时的，也就是说在这个acquiredLocks集合内的多把锁中，第一把获取的锁剩余的有效期一定比最后一把获取的锁的剩余的有效期短。前几个获取到锁的有效期已经消耗了很长时间，这样就有可能会出现有的锁释放了，有的锁没释放。为了避免这种情况，当所有锁都拿到了之后，重新给每把锁的有效期设置为一样的。

在expireAsync方法中，我们可以看到这里其实是使用Lua脚本为所有的锁设置过期时间。

那为什么只有在leaseTime不等于-1的时候才用expireAsync方法确保每把锁的有效期设置为一样的呢？是因为，如果过期时间leaseTime等于-1意味着没有自己指定了锁的过期时间，那上面获取锁的时候会走看门狗机制，有了看门狗机制，所有锁的有效期会自动去续期，不需要进行额外处理了。所以建议大家一般不要设置这个释放时间，让代码直接走看门狗机制。

最终，return true结束。

关于锁的释放

看完加锁逻辑，锁释放就更容易理解了。

我们发现这个释放锁更简单，就是遍历所有的锁资源，然后调用其自身的释放锁方法，进行释放，同步等待所有的资源释放完毕，结束运行。

总结

总结一下，

  1. 不可重入Redis分布式锁
    • 原理：利用SETNX的互斥性；利用EX避免死锁；释放锁时判断线程标识
    • 缺陷：不可重入、无法重试、锁超时失效
  2. 可重入Redis分布式锁
    • 原理：利用Hash结构，记录线程标识与重入次数；利用WatchDog延续锁时间；利用信号量控制锁重试等待
      • ​​​​可重入：利用hash结构记录线程id和重入次数
      • 可重试：利用信号量和PubSub功能实现等待、唤醒，获取锁失败的重试机制
      • 超时续约：利用watchDog，每隔一段时间（releaseTime/3），重置超时时间 
    • 缺陷：Redis宕机引起锁失效问题
  3. Redisson的multiLock
    • 原理：多个独立的Redis节点，必须在所有节点都获取重入锁，才算获取锁成功

版权声明:

本网仅为发布的内容提供存储空间，不对发表、转载的内容提供任何形式的保证。凡本网注明“来源：XXX网络”的作品，均转载自其它媒体，著作权归作者所有，商业转载请联系作者获得授权，非商业转载请注明出处。

我们尊重并感谢每一位作者，均已注明文章来源和作者。如因作品内容、版权或其它问题，请及时与我们联系，联系邮箱：809451989@qq.com，投稿邮箱：809451989@qq.com

最新新闻

热搜词

MySQL 常用的存储引擎-MySQL 常用的存储引擎.png PNG宽高CRC爆破 SpringBoot电商解决方案：星之语明星周边产品销售网站 SQL常用操作 国内外10大知名项目进度跟踪工具对比分析 网络编程(王铭东老师)笔记

声明：本站所有新闻及新闻图片来源于其他网站，如有侵权，请及时联系我们！

客户服务 | 关于我们 | 版权声明

版权所有：

Copyright 2024 尧图网 All Rights Reserved.QQ:809451989