本篇内容介绍了“Redisson RedLock红锁加锁实现过程及原理是什么”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
一、主从redis架构中分布式锁存在的问题
线程A从主redis中请求一个分布式锁，获取锁成功；
从redis准备从主redis同步锁相关信息时，主redis突然发生宕机，锁丢失了；
触发从redis升级为新的主redis；
线程B从继任主redis的从redis上申请一个分布式锁，此时也能获取锁成功；
导致，同一个分布式锁，被两个客户端同时获取，没有保证独占使用特性；
为了解决这个问题，redis引入了红锁的概念。
二、红锁算法原理
需要准备多台redis实例，这些redis实例指的是完全互相独立的Redis节点，这些节点之间既没有主从，也没有集群关系。客户端申请分布式锁的时候，需要向所有的redis实例发出申请，只有超过半数的redis实例报告获取锁成功，才能算真正获取到锁。
具体的红锁算法主要包括如下步骤：
应用程序获取当前系统时间（单位是毫秒）；
应用程序使用相同的key、value依次尝试从所有的redis实例申请分布式锁，这里获取锁的尝试时间要远远小于锁的超时时间，防止某个master Down了，我们还在不断的获取锁，而被阻塞过长的时间；
只有超过半数的redis实例反馈获取锁成功，并且获取锁的总耗时小于锁的超时时间，才认为锁获取成功；
如果锁获取成功了，锁的超时时间就是最初的锁超时时间减去获取锁的总耗时时间；
如果锁获取失败了，不管是因为获取成功的redis节点没有过半，还是因为获取锁的总耗时超过了锁的超时时间，都会向已经获取锁成功的redis实例发出删除对应key的请求，去释放锁；
三、红锁算法的使用
在Redisson框架中，实现了红锁的机制，Redisson的RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例。当红锁中超过半数的RLock加锁成功后，才会认为加锁是成功的，这就提高了分布式锁的高可用。
使用的步骤如下：引入Redisson的maven依赖
<!-- JDK 1.8+ compatible --><dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.9.0</version></dependency>编写单元测试：
@Testpublic void testRedLock() { Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379"); RedissonClient client1 = Redisson.create(config); RLock lock1 = client1.getLock("lock1"); RLock lock2 = client1.getLock("lock2"); RLock lock3 = client1.getLock("lock3"); RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3); try { // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁 boolean res = redLock.tryLock(100, 10, TimeUnit.SECONDS); if (res) { //成功获得锁，在这里处理业务 System.out.println("成功获取到锁..."); } } catch (Exception e) { throw new RuntimeException("aquire lock fail"); } finally { // 无论如何, 最后都要解锁 redLock.unlock(); }}四、红锁加锁流程
RedissonRedLock红锁继承自RedissonMultiLock联锁，简单介绍一下联锁：
基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例，所有的锁都上锁成功才算成功。
RedissonRedLock的加锁、解锁代码都是使用RedissonMultiLock中的方法，只是其重写了一些方法，如：
failedLocksLimit()：允许加锁失败节点个数限制。在RedissonRedLock中，必须超过半数加锁成功才能算成功，其实现为：
protected int failedLocksLimit() { return locks.size() - minLocksAmount(locks);}protected int minLocksAmount(final List<RLock> locks) { // 最小的获取锁成功数：n/2 + 1。 过半机制 return locks.size()/2 + 1;}在RedissonMultiLock中，则必须全部都加锁成功才算成功，所以允许加锁失败节点个数为0，其实现为：
protected int failedLocksLimit() { return 0;}接下来，我们以tryLock()方法为例，详细分析红锁是如何加锁的，具体代码如下：
org.redisson.RedissonMultiLock#tryLock(long, long, java.util.concurrent.TimeUnit)
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {// try {// return tryLockAsync(waitTime, leaseTime, unit).get();// } catch (ExecutionException e) {// throw new IllegalStateException(e);// } long newLeaseTime = -1; if (leaseTime > 0) { if (waitTime > 0) { newLeaseTime = unit.toMillis(waitTime)*2; } else { newLeaseTime = unit.toMillis(leaseTime); } } // 获取当前系统时间，单位：毫秒 long time = System.currentTimeMillis(); long remainTime = -1; if (waitTime > 0) { remainTime = unit.toMillis(waitTime); } long lockWaitTime = calcLockWaitTime(remainTime); // 允许加锁失败节点个数限制（N - ( N / 2 + 1 )） // 假设有三个redis节点，则failedLocksLimit = 1 int failedLocksLimit = failedLocksLimit(); // 存放调用tryLock()方法加锁成功的那些redis节点 List<RLock> acquiredLocks = new ArrayList<>(locks.size()); // 循环所有节点，通过EVAL命令执行LUA脚本进行加锁 for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) { // 获取到其中一个redis实例 RLock lock = iterator.next(); String lockName = lock.getName(); System.out.println("lockName = " + lockName + "正在尝试加锁..."); boolean lockAcquired; try { // 未指定锁超时时间和获取锁等待时间的情况 if (waitTime <= 0 && leaseTime <= 0) { // 调用tryLock()尝试加锁 lockAcquired = lock.tryLock(); } else { // 指定了超时时间的情况,重新计算获取锁的等待时间 long awaitTime = Math.min(lockWaitTime, remainTime); // 调用tryLock()尝试加锁 lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS); } } catch (RedisResponseTimeoutException e) { // 如果抛出RedisResponseTimeoutException异常，为了防止加锁成功，但是响应失败，需要解锁所有节点 unlockInner(Arrays.asList(lock)); // 表示获取锁失败 lockAcquired = false; } catch (Exception e) { // 表示获取锁失败 lockAcquired = false; } if (lockAcquired) { // 如果当前redis节点加锁成功，则加入到acquiredLocks集合中 acquiredLocks.add(lock); } else { // 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)）, 如果已经到达，就认定最终申请锁失败，则没有必要继续从后面的节点申请了。因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功 if (locks.size() - acquiredLocks.size() == failedLocksLimit()) { break; } if (failedLocksLimit == 0) { unlockInner(acquiredLocks); if (waitTime <= 0) { return false; } failedLocksLimit = failedLocksLimit(); acquiredLocks.clear(); // reset iterator while (iterator.hasPrevious()) { iterator.previous(); } } else { failedLocksLimit--; } } // 计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false if (remainTime > 0) { // remainTime: 锁剩余时间，这个时间是某个客户端向所有redis节点申请获取锁的总等待时间, 获取锁的中耗时时间不能大于这个时间。 // System.currentTimeMillis() - time: 这个计算出来的就是当前redis节点获取锁消耗的时间 remainTime -= System.currentTimeMillis() - time; // 重置time为当前时间，因为下一次循环的时候，方便计算下一个redis节点获取锁消耗的时间 time = System.currentTimeMillis(); // 锁剩余时间减到0了，说明达到最大等待时间，加锁超时，认为获取锁失败，需要对成功加锁集合 acquiredLocks 中的所有锁执行锁释放 if (remainTime <= 0) { unlockInner(acquiredLocks); // 直接返回false，获取锁失败 return false; } } } if (leaseTime > 0) { // 重置锁过期时间 acquiredLocks.stream() .map(l -> (RedissonBaseLock) l) .map(l -> l.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS)) .forEach(f -> f.toCompletableFuture().join()); } // 如果逻辑正常执行完则认为最终申请锁成功，返回true return true;}从源码中可以看到，红锁的加锁，其实就是循环所有加锁的节点，挨个执行LUA脚本加锁，对于加锁成功的那些节点，会加入到acquiredLocks集合中保存起来；如果加锁失败的话，则会判断已经申请锁失败的节点是否已经到达允许加锁失败节点个数限制 （N-(N/2+1)）, 如果已经到达，就认定最终申请锁失败，则没有必要继续从后面的节点申请了。
并且，每个节点执行完tryLock()尝试获取锁之后，无论是否获取锁成功，都会判断目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，需要对成功加锁集合 acquiredLocks 中的所有锁执行锁释放，然后返回false。
五、RedLock算法问题
持久化问题
假设一共有5个Redis节点：A, B, C, D, E：
客户端1成功锁住了A, B, C，获取锁成功，但D和E没有锁住。
节点C崩溃重启了，但客户端1在C上加的锁没有持久化下来，丢失了。
节点C重启后，客户端2锁住了C, D, E，获取锁成功。
这样，客户端1和客户端2同时获得了锁（针对同一资源）。
客户端长时间阻塞，导致获得的锁释放，访问的共享资源不受保护的问题。
Redlock算法对时钟依赖性太强， 若某个节点中发生时间跳跃（系统时间戳不正确），也可能会引此而引发锁安全性问题。
“Redisson RedLock红锁加锁实现过程及原理是什么”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注编程网网站，小编将为大家输出更多高质量的实用文章！
