在 《精尽 Redisson 源码分析 —— 可重入分布式锁 ReentrantLock》 中，艿艿臭长臭长的分享了 Redisson 是如何实现可重入的 ReentrantLock 锁，一切看起来很完美，我们能够正确的加锁，也能正确的释放锁。但是，我们来看一个 Redis 主从结构下的示例，Redis 分布式锁是如何失效的：

  • 1、客户端 A 从 Redis Master 获得到锁 anylock 。
  • 2、在 Redis Master 同步 anylock 到 Redis Slave 之前，Master 挂了。
  • 3、Redis Slave 晋升为新的 Redis Master 。
  • 4、客户端 B 从新的 Redis Master 获得到锁 anylock 。

此时，客户端 A 和 B 同时持有 anylock 锁，已经失效。当然，这个情况是极小概率事件。主要看胖友业务对分布式锁可靠性的诉求。

在 Redis 分布式锁存在失效的问题，Redis 的作者 Antirez 大神提出了红锁 RedLock 的想法。我们来看看它的描述。

FROM 《Redis 文档 —— Redis 分布式锁》

在 Redis 的分布式环境中，我们假设有 N 个 Redis master 。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。之前我们已经描述了在 Redis 单实例下怎么安全地获取和释放锁。我们确保将在每 N 个实例上使用此方法获取和释放锁。在这个样例中，我们假设有 5 个 Redis master 节点，这是一个比较合理的设置，所以我们需要在 5 台机器上面或者 5 台虚拟机上面运行这些实例，这样保证他们不会同时都宕掉。

为了取到锁，客户端应该执行以下操作:

  • 1、获取当前 Unix 时间，以毫秒为单位。
  • 2、依次尝试从 N 个实例，使用相同的 key 和随机值获取锁。在步骤 2 ，当向 Redis 设置锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为 10 秒，则超时时间应该在 5-50 毫秒之间。这样可以避免服务器端 Redis 已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试另外一个 Redis 实例。
  • 3、客户端使用当前时间减去开始获取锁时间（步骤 1 记录的时间）就得到获取锁使用的时间。当且仅当从大多数（这里是 3 个节点）的 Redis 节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
  • 4、如果取到了锁，key 的真正有效时间等于有效时间减去获取锁所使用的时间（步骤 3 计算的结果）。
  • 5、如果因为某些原因，获取锁失败（没有在至少 N/2 + 1 个 Redis 实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的 Redis 实例上进行解锁（即便某些 Redis 实例根本就没有加锁成功）。

释放锁：

  • 1、释放锁比较简单，向所有的 Redis 实例发送释放锁命令即可，不用关心之前有没有从 Redis 实例成功获取到锁.

可能一看内容这么长，略微有点懵逼。重点理解，需要至少在 N/2 + 1 Redis 节点获得锁成功。这样，即使出现某个 Redis Master 未同步锁信息到 Redis Slave 节点之前，突然挂了，也不容易出现多个客户端获得相同锁，因为需要至少在 N/2 + 1 Redis 节点获得锁成功。。

当然，极端情况下也有。我们以 3 个 Redis Master 节点举例子：

  • 1、客户端 A 从 3 个 Redis Master 获得到锁 anylock 。
  • 2、2 个 Redis Master 同步 anylock 到 Redis Slave 之前，Master 都挂了。
  • 3、2 个 Redis Slave 晋升为新的 Redis Master 。
  • 4、客户端 B 从新的 Redis Master 获得到锁 anylock 。

理论来说，出现 2 个 Redis Master 都挂了，并且数据都未同步到 Redis Slave 的情况，已经是小概率的事件。当然，哈哈哈哈，我们就是可爱的 “杠精”，就是要扣一扣这个边界情况。

同时，我们也可以发现，在时候用 RedLock 的时候，Redis Master 越多，集群的可靠性就越高，性能也会越低。😈 架构设计中，从来没有银弹。我们想要得到更高的可靠性，往往需要失去一定的性能。

对了，有一点要注意，N 个 Redis Master 要毫无关联的。例如说，任一一个 Redis Master 都不能在同一个 Redis Cluster 中。再如下图，就是一个符合条件的：

FROM 《慢谈 Redis 实现分布式锁 以及 Redisson 源码解析》

精尽 Redisson 源码分析 —— 可靠分布式锁 RedLock - 图1
Redis Master 示例

当然，推荐 N 是奇数个，因为 N / 2 + 1 嘛，哈哈。

推荐胖友阅读如下三篇文章，更进一步了解 RedLock ：

org.redisson.RedissonRedLock ，继承自联锁 RedissonMultiLock ，Redisson 对 RedLock 的实现类。代码如下：

public class RedissonRedLock extends RedissonMultiLock {

public RedissonRedLock(RLock… locks) {
super(locks);
}

@Override
protected int failedLocksLimit() {
return locks.size() - minLocksAmount(locks);
}

protected int minLocksAmount(final List locks) {
return locks.size()/2 + 1;
}

@Override
protected long calcLockWaitTime(long remainTime) {
return Math.max(remainTime / locks.size(), 1);
}

@Override
public void unlock() {
unlockInner(locks);
}

}

  • RedissonMultiLock ，联锁，正如其名字 “联”，可以将多个 RLock 锁关联成一个联锁。使用示例如下：
    RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);
    lock.lock(10, TimeUnit.SECONDS);
    boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
    …
    lock.unlock();

  • RedissonRedLock ，是一个特殊的联锁，加锁时无需所有的 RLock 都成功，只需要满足 N / 2 + 1 个 RLock 即可。使用示例如下：
    RLock lock1 = redissonInstance1.getLock(“lock1”);
    RLock lock2 = redissonInstance2.getLock(“lock2”);
    RLock lock3 = redissonInstance3.getLock(“lock3”);
    RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);
    lock.lock();
    …
    lock.unlock();

  • RedissonRedLock 构造方法，创建时，传入多个 RLock 对象。一般来说，每个 RLock 对应到一个 Redis Master 节点上。

  • #failedLocksLimit() 方法，允许失败加锁的数量。从 #minLocksAmount(final List<RLock> locks) 方法上，我们已经看到 N / 2 + 1 的要求。

  • #calcLockWaitTime(long remainTime) 方法，计算每次获得 RLock 的锁的等待时长。目前的计算规则是，总的等待时间 remainTime 平均分配到每个 RLock 上。

  • #unlock() 方法，解锁时，需要所有 RLock 都进行解锁。

org.redisson.RedissonMultiLock ，实现 RLock 接口，Redisson 对联锁 MultiLock 的实现类。

3.1 构造方法

final List locks = new ArrayList<>();

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

3.2 failedLocksLimit

#failedLocksLimit() 方法，允许失败加锁的数量。代码如下：

protected int failedLocksLimit() {
return 0;
}

默认返回值是 0 ，也就是必须所有 RLock 都加锁成功。

在 RedissonRedLock 中，会重写该方法。

3.3 calcLockWaitTime

#failedLocksLimit(long remainTime) 方法，计算每次获得 RLock 的锁的等待时长。代码如下：

protected long calcLockWaitTime(long remainTime) {
return remainTime;
}

默认直接返回 remainTime ，也就是说，每次获得 RLock 的锁的等待时长都是 remainTime 。

在 RedissonRedLock 中，会重写该方法。

3.4 tryLock

#tryLock(long waitTime, long leaseTime, TimeUnit unit) 方法，同步加锁，并返回加锁是否成功。代码如下：

@Override
public boolean tryLock() {
try {

return tryLock(-1, -1, null);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}

@Override
public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
return tryLock(waitTime, -1, unit);
}

1: @Override
2: public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
3:
4: long newLeaseTime = -1;
5: if (leaseTime != -1) {
6: if (waitTime == -1) {
7: newLeaseTime = unit.toMillis(leaseTime);
8: } else {
9: newLeaseTime = unit.toMillis(waitTime) * 2;
10: }
11: }
12:
13: long time = System.currentTimeMillis();
14:
15: long remainTime = -1;
16: if (waitTime != -1) {
17: remainTime = unit.toMillis(waitTime);
18: }
19:
20: long lockWaitTime = calcLockWaitTime(remainTime);
21:
22:
23: int failedLocksLimit = failedLocksLimit();
24:
25: List acquiredLocks = new ArrayList<>(locks.size());
26:
27: for (ListIterator iterator = locks.listIterator(); iterator.hasNext();) {
28:
29: RLock lock = iterator.next();
30: boolean lockAcquired;
31: try {
32:
33: if (waitTime == -1 && leaseTime == -1) {
34: lockAcquired = lock.tryLock();
35:
36: } else {
37: long awaitTime = Math.min(lockWaitTime, remainTime);
38: lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
39: }
40: } catch (RedisResponseTimeoutException e) {
41:
42: unlockInner(Collections.singletonList(lock));
43:
44: lockAcquired = false;
45: } catch (Exception e) {
46:
47: lockAcquired = false;
48: }
49:
50:
51: if (lockAcquired) {
52: acquiredLocks.add(lock);
53: } else {
54:
55: if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
56: break;
57: }
58:
59:
60: if (failedLocksLimit == 0) {
61:
62: unlockInner(acquiredLocks);
63:
64: if (waitTime == -1) {
65: return false;
66: }
67:
68:
69: failedLocksLimit = failedLocksLimit();
70:
71: acquiredLocks.clear();
72:
73:
74: while (iterator.hasPrevious()) {
75: iterator.previous();
76: }
77:
78: } else {
79: failedLocksLimit—;
80: }
81: }
82:
83:
84: if (remainTime != -1) {
85: remainTime -= System.currentTimeMillis() - time;
86:
87: time = System.currentTimeMillis();
88:
89: if (remainTime <= 0) {
90: unlockInner(acquiredLocks);
91: return false;
92: }
93: }
94: }
95:
96:
97: if (leaseTime != -1) {
98:
99: List futures = new ArrayList<>(acquiredLocks.size());
100: for (RLock rLock : acquiredLocks) {
101: RFuture future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
102: futures.add(future);
103: }
104:
105:
106: for (RFuture rFuture : futures) {
107: rFuture.syncUninterruptibly();
108: }
109: }
110:
111:
112: return true;
113: }

  • 超 100 行代码，是不是有点慌？！核心逻辑是，首先从 locks 数组中逐个获得锁（第 27 至 94 行），然后统一设置每个锁的过期时间（第 96 至 10 9 行）。当然，还是有很多细节，我们一点点来看。
  • 第 3 至 11 行：计算新的锁的时长 newLeaseTime 。注意，newLeaseTime 是用于遍历 locks 数组来获得锁设置的锁时长，最终在第 96 至 109 行的代码中，会设置真正的 leaseTime 锁的时长。整个的计算规则，看下艿艿添加的注释。
  • 第 14 至 18 行：计算剩余等待锁的时间 remainTime 。
  • 第 20 行：调用 #calcLockWaitTime(long remainTime) 方法，计算获得每个锁的等待时间。在 「2. RedissonRedLock」 中，我们已经看到，是 remainTime 进行平均分配。
  • 第 25 行：已经获得到锁的数组 acquiredLocks 。
  • 下面，我们分成阶段一（加锁）和阶段二（设置锁过期时间）来抽丝剥茧。
  • 【阶段一】第 26 至 94 行：遍历 RLock 数组，逐个获得锁。
  • 第 32 至 39 行：根据条件，调用对应的 RLock#tryLock(...) 方法，获得锁。一般情况下，RedissonRedLock 搭配 RedissonLock 使用，这块我们已经在 《精尽 Redisson 源码分析 —— 可重入分布式锁 ReentrantLock》 有个详细的解析了。
  • 第 40 至 44 行：如果发生响应 RedisResponseTimeoutException 超时异常时，因为无法确定实际是否获得到锁，所以直接调用 #unlockInner(Collection<RLock> locks) 方法，释放当前 RLock 。
  • 第 50 至 52 行：如果获得成功，则添加到 acquiredLocks 数组中。
  • 【重要】第 54 至 57 行：如果已经到达最少需要获得锁的数量，则直接 break 。例如说，RedLock 只需要获得 N / 2 + 1 把。
  • 第 77 至 80 行：failedLocksLimit 减一。

  • 第 59 至 76 行：当已经没有允许失败的数量，则进行相应的处理。

    • 第 62 行：因为已经失败了，所以调用 #unlockInner(Collection<RLock> locks) 方法，释放所有已经获得到的锁们。
    • 第 63 至 66 行：如果未设置阻塞时间，直接返回 false 加锁失败。因为是 tryLock 方法，只是尝试加锁一次，不会无限重试。
    • 第 67 至 76 行：重置整个获得锁的过程，在剩余的时间里，重新来一遍。因为已设置了阻塞时间，必须得用完！
  • 第 84 至 93 行：计算剩余时间 remainTime 。如果没有剩余的时间，意味着已经超时，则 调用 #unlockInner(Collection<RLock> locks) 方法，释放所有加载成功的锁，并返回 false 加锁失败。
  • 【阶段二】第 96 至 109 行：如果设置了锁的过期时间 leaseTime ，则重新设置每个锁的过期时间。
  • 第 98 至 103 行：遍历 acquiredLocks 数组，创建异步设置过期时间的 Future 。
  • 第 105 至 108 行：阻塞等待所有 futures 完成。如果任一一个 Future 执行失败，则会抛出异常。

3.5 tryLockAsync

#tryLockAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) 方法，异步加锁，并返回加锁是否成功。代码如下：

@Override
public RFuture tryLockAsync(long waitTime, long leaseTime, TimeUnit unit) {
return tryLockAsync(waitTime, leaseTime, unit, Thread.currentThread().getId());
}

@Override
public RFuture tryLockAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {

RPromise result = new RedissonPromise();

LockState state = new LockState(waitTime, leaseTime, unit, threadId);

state.tryAcquireLockAsync(locks.listIterator(), result);

return result;
}

  • 这个的逻辑在 <X> 处，调用 LockState#tryAcquireLockAsync(ListIterator<RLock> iterator, RPromise<Boolean> result) 方法，发起异步加锁。

🔥 LockState 是 RedissonMultiLock 的内部类，实现异步加锁的逻辑。构造方法如下：

class LockState {

private final long newLeaseTime;
private final long lockWaitTime;
private final List acquiredLocks;
private final long waitTime;
private final long threadId;
private final long leaseTime;
private final TimeUnit unit;

private long remainTime;
private long time = System.currentTimeMillis();
private int failedLocksLimit;

LockState(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
this.waitTime = waitTime;
this.leaseTime = leaseTime;
this.unit = unit;
this.threadId = threadId;

if (leaseTime != -1) {
if (waitTime == -1) {
newLeaseTime = unit.toMillis(leaseTime);
} else {
newLeaseTime = unit.toMillis(waitTime) * 2;
}
} else {
newLeaseTime = -1;
}

remainTime = -1;
if (waitTime != -1) {
remainTime = unit.toMillis(waitTime);
}

lockWaitTime = calcLockWaitTime(remainTime);

failedLocksLimit = failedLocksLimit();

acquiredLocks = new ArrayList<>(locks.size());
}

}

🔥 LockState#tryAcquireLockAsync(ListIterator<RLock> iterator, RPromise<Boolean> result) 方法，发起异步加锁。代码如下：

整体逻辑，和 「3.4 tryLock(long waitTime, long leaseTime, TimeUnit unit)」 方法的逻辑基本一致，所以艿艿就不啰嗦详细说，而是告诉它们的对等关系。

1: void tryAcquireLockAsync(ListIterator iterator, RPromise result) {
2:
3: if (!iterator.hasNext()) {
4: checkLeaseTimeAsync(result);
5: return;
6: }
7:
8:
9: RLock lock = iterator.next();
10:
11: RPromise lockAcquiredFuture = new RedissonPromise();
12:
13: if (waitTime == -1 && leaseTime == -1) {
14: lock.tryLockAsync(threadId)
15: .onComplete(new TransferListener(lockAcquiredFuture));
16:
17: } else {
18: long awaitTime = Math.min(lockWaitTime, remainTime);
19: lock.tryLockAsync(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS, threadId)
20: .onComplete(new TransferListener(lockAcquiredFuture));
21: }
22:
23: lockAcquiredFuture.onComplete((res, e) -> {
24:
25: boolean lockAcquired = false;
26: if (res != null) {
27: lockAcquired = res;
28: }
29:
30:
31: if (e instanceof RedisResponseTimeoutException) {
32: unlockInnerAsync(Collections.singletonList(lock), threadId);
33: }
34:
35:
36: if (lockAcquired) {
37: acquiredLocks.add(lock);
38: } else {
39:
40: if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
41: checkLeaseTimeAsync(result);
42: return;
43: }
44:
45:
46: if (failedLocksLimit == 0) {
47:
48: unlockInnerAsync(acquiredLocks, threadId).onComplete((r, ex) -> {
49:
50: if (ex != null) {
51: result.tryFailure(ex);
52: return;
53: }
54:
55:
56: if (waitTime == -1) {
57: result.trySuccess(false);
58: return;
59: }
60:
61:
62:
63: failedLocksLimit = failedLocksLimit();
64:
65: acquiredLocks.clear();
66:
67:
68: while (iterator.hasPrevious()) {
69: iterator.previous();
70: }
71:
72:
73: checkRemainTimeAsync(iterator, result);
74: });
75: return;
76: } else {
77: failedLocksLimit—;
78: }
79: }
80:
81:
82: checkRemainTimeAsync(iterator, result);
83: });
84: }

  • 第 2 至 6 行：如果迭代 iterator 的尾部，则调用 #checkLeaseTimeAsync(RPromise<Boolean> result) 方法，重新设置每个锁的过期时间。代码如下：
    private void checkLeaseTimeAsync(RPromise result) {
    if (leaseTime != -1) {
    AtomicInteger counter = new AtomicInteger(acquiredLocks.size());
    for (RLock rLock : acquiredLocks) {
    RFuture future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);
    future.onComplete((res, e) -> {
    if (e != null) {
    result.tryFailure(e);
    return;
    }
    if (counter.decrementAndGet() == 0) {
    result.trySuccess(true);
    }
    });
    }
    return;
    }
    result.trySuccess(true);
    }

  • 第 8 至 79 行：对标到 「3.4 tryLock(long waitTime, long leaseTime, TimeUnit unit)」 的第 28 至 81 行。

  • 第 73 和 82 行：调用 #checkRemainTimeAsync(ListIterator<RLock> iterator, RPromise<Boolean> result) 方法，校验剩余时间是否足够。代码如下：
    private void checkRemainTimeAsync(ListIterator iterator, RPromise result) {
    if (remainTime != -1) {
    remainTime += -(System.currentTimeMillis() - time);
    time = System.currentTimeMillis();
    if (remainTime <= 0) {
    unlockInnerAsync(acquiredLocks, threadId).onComplete((res, e) -> {
    if (e != null) {
    result.tryFailure(e);
    return;
    }
    result.trySuccess(false);
    });
    return;
    }
    }
    tryAcquireLockAsync(iterator, result);
    }

😈 总的来说，还是蛮简单的不是，哈哈哈哈。

3.6 lock

#lockInterruptibly(long leaseTime, TimeUnit unit) 方法，同步加锁，不返回加锁是否成功。代码如下：

@Override
public void lock() {
try {
lockInterruptibly();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

@Override
public void lock(long leaseTime, TimeUnit unit) {
try {
lockInterruptibly(leaseTime, unit);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}

@Override
public void lockInterruptibly() throws InterruptedException {
lockInterruptibly(-1, null);
}

1: @Override
2: public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
3:
4: long baseWaitTime = locks.size() * 1500;
5: long waitTime = -1;
6: if (leaseTime == -1) {
7: waitTime = baseWaitTime;
8: } else {
9: leaseTime = unit.toMillis(leaseTime);
10: waitTime = leaseTime;
11: if (waitTime <= 2000) {
12: waitTime = 2000;
13: } else if (waitTime <= baseWaitTime) {
14: waitTime = ThreadLocalRandom.current().nextLong(waitTime / 2, waitTime);
15: } else {
16: waitTime = ThreadLocalRandom.current().nextLong(baseWaitTime, waitTime);
17: }
18: }
19:
20:
21: while (true) {
22: if (tryLock(waitTime, leaseTime, TimeUnit.MILLISECONDS)) {
23: return;
24: }
25: }
26: }

3.7 lockAsync

#lockAsync(long leaseTime, TimeUnit unit, long threadId) 方法，异步加锁，不返回加锁是否成功。代码如下：

@Override
public RFuture lockAsync(long leaseTime, TimeUnit unit) {
return lockAsync(leaseTime, unit, Thread.currentThread().getId());
}

1: @Override
2: public RFuture lockAsync(long leaseTime, TimeUnit unit, long threadId) {
3:
4: long baseWaitTime = locks.size() * 1500;
5: long waitTime;
6: if (leaseTime == -1) {
7: waitTime = baseWaitTime;
8: } else {
9: leaseTime = unit.toMillis(leaseTime);
10: waitTime = leaseTime;
11: if (waitTime <= 2000) {
12: waitTime = 2000;
13: } else if (waitTime <= baseWaitTime) {
14: waitTime = ThreadLocalRandom.current().nextLong(waitTime / 2, waitTime);
15: } else {
16: waitTime = ThreadLocalRandom.current().nextLong(baseWaitTime, waitTime);
17: }
18: }
19:
20:
21: RPromise result = new RedissonPromise();
22:
23: tryLockAsync(threadId, leaseTime, TimeUnit.MILLISECONDS, waitTime, result);
24: return result;
25: }

  • 第 3 至 18 行：计算 waitTime 时间。和 #lockInterruptibly(long leaseTime, TimeUnit unit) 方法，看到的逻辑是一致的。

  • 第 23 行：调用 #tryLockAsync(long threadId, long leaseTime, TimeUnit unit, long waitTime, RPromise<Void> result) 方法，执行异步加锁。代码如下：
    protected void tryLockAsync(long threadId, long leaseTime, TimeUnit unit, long waitTime, RPromise result) {
    tryLockAsync(waitTime, leaseTime, unit, threadId).onComplete((res, e) -> {
    if (e != null) {
    result.tryFailure(e);
    return;
    }
    if (res) {
    result.trySuccess(null);
    } else {
    tryLockAsync(threadId, leaseTime, unit, waitTime, result);
    }
    });
    }

3.8 unlock

#unlock() 方法，同步解锁。代码如下：

@Override
public void unlock() {

List futures = new ArrayList<>(locks.size());

for (RLock lock : locks) {
futures.add(lock.unlockAsync());
}

for (RFuture future : futures) {
future.syncUninterruptibly();
}
}

在 RedissonMultiLock 类中，存在一个 #unlockInner(Collection<RLock> locks) 方法，同步解锁指定 RLock 数组。代码如下：

protected void unlockInner(Collection locks) {

List futures = new ArrayList<>(locks.size());

for (RLock lock : locks) {
futures.add(lock.unlockAsync());
}

for (RFuture unlockFuture : futures) {
unlockFuture.awaitUninterruptibly();
}
}

在 RedissonMultiLock 类中，存在一个 #unlockInner(Collection<RLock> locks) 方法，异步解锁指定 RLock 数组。代码如下：

protected RFuture unlockInnerAsync(Collection locks, long threadId) {

if (locks.isEmpty()) {
return RedissonPromise.newSucceededFuture(null);
}

RPromise result = new RedissonPromise();

AtomicInteger counter = new AtomicInteger(locks.size());

for (RLock lock : locks) {
lock.unlockAsync(threadId).onComplete((res, e) -> {

if (e != null) {
result.tryFailure(e);
return;
}

if (counter.decrementAndGet() == 0) {
result.trySuccess(null);
}
});
}
return result;
}

3.9 未实现的方法

在 RedissonMultiLock 中，因为一些方法暂时没必要实现，所以就都未提供。如下：

@Override
public Condition newCondition() {
throw new UnsupportedOperationException();
}

@Override
public RFuture forceUnlockAsync() {
throw new UnsupportedOperationException();
}

@Override
public RFuture getHoldCountAsync() {
throw new UnsupportedOperationException();
}

@Override
public String getName() {
throw new UnsupportedOperationException();
}

@Override
public boolean forceUnlock() {
throw new UnsupportedOperationException();
}

@Override
public boolean isLocked() {
throw new UnsupportedOperationException();
}

@Override
public RFuture isLockedAsync() {
throw new UnsupportedOperationException();
}

@Override
public boolean isHeldByThread(long threadId) {
throw new UnsupportedOperationException();
}

@Override
public boolean isHeldByCurrentThread() {
throw new UnsupportedOperationException();
}

@Override
public int getHoldCount() {
throw new UnsupportedOperationException();
}

@Override
public RFuture remainTimeToLiveAsync() {
throw new UnsupportedOperationException();
}

@Override
public long remainTimeToLive() {
throw new UnsupportedOperationException();
}

一开始，以为 RedLock 红锁的代码会比较复杂，所以在撸这块的源码时，有点懵逼。一度计划，准备花小 1 天的时间来研究和输出这篇博客。结果发现，竟然是个纸老虎，哈哈哈。

😈 所以把，碰到任何源码，都不要怂。该肝就肝！

爽，在 2019-10-05 的 01:30 写完了这篇博客，美滋滋。
http://svip.iocoder.cn/Redisson/RedLock/