对于redis 集群模式下，若master 的锁还没有同步给slave，这时 master 挂掉，然后哨兵选举出新的master，
由于新的 master 并没有同步到锁，所以这个时候其他的线程仍然能获取到锁。因此独占式在一定条件下是会失效的。
这个观点在另外几篇参考的文章中也有提到，个人也比较赞同，因此在此写个笔记。

通过分析底层代码，当锁没有设置自动释放时间才会启用看门狗线程的。
所以我们要预设置过期时间的话最好还是先预估任务的实际执行时间再进行取值为妙...

齿轮有两个属性：

task：被执行的任务
bound：当bound = 0 时 task才会被执行，当bound > 0 时，指针每过一次bound - 1 直至为0 。
eg：如果你想31s后执行任务，那么bound应该等于3，齿轮处于第7个位置上。因为3*8+7=31。

4.3：解锁->unlock

底层源码：

①：若当前线程并没有持有锁，则返回nil。
②：当前线程持有锁，则对value-1，拿到-1之后的vlaue。
③：value>0，以毫秒为单位返回剩下的过期时间。（保证可重入）
④：value<=0，则 ad0 对key进行删除操作，return 1 (方法返回 true)。然后进行redis-pub指令。

5：获取锁失败

5.1：关系类图

这块儿比较麻烦，先给一下比较重要的类图吧...

5.2：订阅事件

没获取到锁线程后面在干嘛？当然要持续等待啦...
先在redis中发布订阅消息，等待用完锁的线程通知我~

看看订阅主要干了些啥，从源码上分析一波

①：PublishSubscribe.subscribe(String entryName, String channelName)源码：

public RFuture<E> subscribe(String entryName, String channelName) {
AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
RPromise<E> newPromise = new RedissonPromise<>();
semaphore.acquire(() -> {
if (!newPromise.setUncancellable()) {
semaphore.release();
return;
}
// 1:判断RedisLockEntry 是否存在
E entry = entries.get(entryName);
if (entry != null) {
entry.acquire();
semaphore.release();
entry.getPromise().onComplete(new TransferListener<E>(newPromise));
return;
}

// 2：创建RedisLockEntry
E value = createEntry(newPromise);
value.acquire();

E oldValue = entries.putIfAbsent(entryName, value);
if (oldValue != null) {
oldValue.acquire();
semaphore.release();
oldValue.getPromise().onComplete(new TransferListener<E>(newPromise));
return;
}
// 3:创建一个监听器，别的线程进行redis-pub命令之后进行调用
RedisPubSubListener<Object> listener = createListener(channelName, value);
// 4:底层交给netty调用redis-sub命令
service.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);
});

return newPromise;
}

②：AsyncSemaphore.acquire(Runnable listener)源码：

public class AsyncSemaphore {
private final AtomicInteger counte
15b0
r;
private final Queue<Runnable> listeners = new ConcurrentLinkedQueue<>();
...

public void acquire(Runnable listener) {
1：将任务假如到阻塞队列
listeners.add(listener);
tryRun();
}

private void tryRun() {
if (counter.get() == 0
|| listeners.peek() == null) {
return;
}
// 2：counter>0时tryRun 才会取出linstener 中的任务进行执行
if (counter.decrementAndGet() >= 0) {
Runnable listener = listeners.poll();
if (listener == null) {
counter.incrementAndGet();
return;
}

if (removedListeners.remove(listener)) {
counter.incrementAndGet();
tryRun();
} else {
listener.run();
}
} else {
counter.incrementAndGet();
}
}
}

③：PubSubLock.createEntry() 源码：

@Override
protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {
return new RedissonLockEntry(newPromise);
}

④：RedisLockEntry 的部分源码：

public class RedissonLockEntry implements PubSubEntry<RedissonLockEntry> {

private int counter;

private final Semaphore latch;
private final RPromise<RedissonLockEntry> promise;
private final ConcurrentLinkedQueue<Runnable> listeners = new ConcurrentLinkedQueue<Runnable>();

public RedissonLockEntry(RPromise<RedissonLockEntry> promise) {
super();
this.latch = new Semaphore(0);
this.promise = promise;
}
...
public Semaphore getLatch() {
return latch;
}
}

⑤：RedisPubSubConnection.subscribe(Codec codec, ChannelName... channels)源码：

public ChannelFuture subscribe(Codec codec, ChannelName... channels) {
for (ChannelName ch : channels) {
this.channels.put(ch, codec);
}
return async(new PubSubMessageDecoder(codec.getValueDecoder()), RedisCommands.SUBSCRIBE, channels);
}

...
private <T, R> ChannelFuture async(MultiDecoder<Object> messageDecoder, RedisCommand<T> command, Object... params) {
RPromise<R> promise = new RedissonPromise<R>();
// io.netty.Channel
return channel.writeAndFlush(new CommandData<T, R>(promise, messageDecoder, null, command, params));
}

给张图可能看起来方便点：

订阅源码总结：
①：并不是每次每次都会创建RedisLockEntry，理论上是：当前应用内一个channel 对应一个RedisLockEntry 实例。
②：subscribe 的底层是基于netty进行操作的，并不是基于RedisTemplate。
③：不是每次subscribe都会执行到netty层，只有当属于该redis-channel的RedisLockEntry 没有实例化时才会调用到netty层。后续线程的只需要执行RedisLockEntry.acquire 操作即可。

6：redis-pub和redis-sub 是如何遥相呼应的？

6.1：Semaphore.tryAcquire(...)

RedisLockEntry 的latch属性为Semaphore

我们看看RedisLock.lock() 源码：

public void onMessage(PubSubMessage message) {
for (RedisPubSubListener<Object> redisPubSubListener : listeners) {
redisPubSubListener.onMessage(message.getChannel(), message.getValue());
}
}

@Override
protected void onMessage(RedissonLockEntry value, Long message) {
if (message.equals(UNLOCK_MESSAGE)) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute != null) {
runnableToExecute.run();
}

// value 就是线程进行subscribe 操作的时候所使用的RedisLockEntry 对象
// 终于执行release操作了，等待锁的线程可以再次获取锁了
value.getLatch().release();
} else if (message.equals(READ_U
56c
NLOCK_MESSAGE)) {
while (true) {
Runnable runnableToExecute = value.getListeners().poll();
if (runnableToExecute == null) {
break;
}
runnableToExecute.run();
}

value.getLatch().release(value.getLatch().getQueueLength());
}
}