分布式锁和Redisson实现

深藏阁楼爱情的钟 2022-05-18 04:36 318阅读 0赞

## 概述 ##

分布式系统有一个著名的理论CAP，指在一个分布式系统中，最多只能同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance）这三项中的两项。所以在设计系统时，往往需要权衡，在CAP中作选择。当然，这个理论也并不一定完美，不同系统对CAP的要求级别不一样，选择需要考虑方方面面。

在微服务系统中，一个请求存在多级跨服务调用，往往需要牺牲强一致性老保证系统高可用，比如通过分布式事务，异步消息等手段完成。但还是有的场景，需要阻塞所有节点的所有线程，对共享资源的访问。比如并发时“超卖”和“余额减为负数”等情况。

本地锁可以通过语言本身支持，要实现分布式锁，就必须依赖中间件，数据库、redis、zookeeper等。

## 分布式锁特性 ##

不管使用什么中间件，有几点是实现分布式锁必须要考虑到的。

1.  **互斥**：互斥好像是必须的，否则怎么叫锁。
2.  **死锁**: 如果一个线程获得锁，然后挂了，并没有释放锁，致使其他节点(线程)永远无法获取锁，这就是死锁。分布式锁必须做到避免死锁。
3.  **性能**: 高并发分布式系统中，线程互斥等待会成为性能瓶颈，需要好的中间件和实现来保证性能。
4.  **锁特性**：考虑到复杂的场景，分布式锁不能只是加锁，然后一直等待。最好实现如Java Lock的一些功能如：锁判断，超时设置，可重入性等。

## Redis实现之Redisson原理 ##

redission实现了JDK中的Lock接口，所以使用方式一样，只是Redssion的锁是分布式的。如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 </td> 
 <td> RLock lock = redisson.getLock("className"); lock.lock(); try { // do sth. }&nbsp;finally { lock.unlock(); } </td> 
 </tr> 
 </tbody> 
</table>

好，Lock主要实现是RedissionLock。  
[![2017-08-19-16-44-27.jpg][]][2017-08-19-16-44-27.jpg]

先来看常用的Lock方法实现。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 </td> 
 <td> @Override public&nbsp;void&nbsp;lock() { try { lockInterruptibly(); }&nbsp;catch (InterruptedException e) { Thread.currentThread().interrupt(); } } @Override public&nbsp;void&nbsp;lockInterruptibly()&nbsp;throws InterruptedException { lockInterruptibly(-1,&nbsp;null); } </td> 
 </tr> 
 </tbody> 
</table>

再看`lockInterruptibly`方法：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 </td> 
 <td> @Override public&nbsp;void&nbsp;lockInterruptibly(long leaseTime, TimeUnit unit)&nbsp;throws InterruptedException { long threadId = Thread.currentThread().getId(); // 获取锁 Long ttl = tryAcquire(leaseTime, unit, threadId); if (ttl ==&nbsp;null) {&nbsp;// 获取成功 return; } &nbsp; // 异步订阅redis chennel RFuture&lt;RedissonLockEntry&gt; future = subscribe(threadId); commandExecutor.syncSubscription(future);&nbsp;// 阻塞获取订阅结果 &nbsp; try { while (true) {// 循环判断知道获取锁 ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl ==&nbsp;null) { break; } &nbsp; // waiting for message if (ttl &gt;=&nbsp;0) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); }&nbsp;else { getEntry(threadId).getLatch().acquire(); } } }&nbsp;finally { unsubscribe(future, threadId);// 取消订阅 } } </td> 
 </tr> 
 </tbody> 
</table>

总结`lockInterruptibly`：获取锁，不成功则订阅释放锁的消息，获得消息前阻塞。得到释放通知后再去循环获取锁。

下面重点看看如何获取锁：Long ttl = tryAcquire(leaseTime, unit, threadId)

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 </td> 
 <td> private Long&nbsp;tryAcquire(long leaseTime, TimeUnit unit,&nbsp;long threadId) { return get(tryAcquireAsync(leaseTime, unit, threadId));// 通过异步获取锁，但get(future)实现同步 } </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 </td> 
 <td> private &lt;T&gt;&nbsp;RFuture&lt;Long&gt;&nbsp;tryAcquireAsync(long leaseTime, TimeUnit unit,&nbsp;final&nbsp;long threadId) { if (leaseTime != -1) {&nbsp;//1 如果设置了超时时间，直接调用 tryLockInnerAsync return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } //2 如果leaseTime==-1，则默认超时时间为30s RFuture&lt;Long&gt; ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG); //3 监听Future，获取Future返回值ttlRemaining(剩余超时时间)，获取锁成功，但是ttlRemaining，则刷新过期时间 ttlRemainingFuture.addListener(new FutureListener&lt;Long&gt;() { @Override public&nbsp;void&nbsp;operationComplete(Future&lt;Long&gt; future)&nbsp;throws Exception { if (!future.isSuccess()) { return; } &nbsp; Long ttlRemaining = future.getNow(); // lock acquired if (ttlRemaining ==&nbsp;null) { scheduleExpirationRenewal(threadId); } } }); return ttlRemainingFuture; } </td> 
 </tr> 
 </tbody> 
</table>

已经在注释中解释了，需要注意的是，此处用到了Netty的Future-listen模型，可以看看我的另一篇对Future的简单讲解：[给Future一个Promise][Future_Promise]。

下面就是最重要的redis获取锁的方法`tryLockInnerAsync`:

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 </td> 
 <td> &lt;T&gt;&nbsp;RFuture&lt;T&gt;&nbsp;tryLockInnerAsync(long leaseTime, TimeUnit unit,&nbsp;long threadId, RedisStrictCommand&lt;T&gt; command) { internalLockLeaseTime = unit.toMillis(leaseTime); return commandExecutor.evalWriteAsync( getName(), LongCodec.INSTANCE, command, "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "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]);", Collections.&lt;Object&gt;singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); } </td> 
 </tr> 
 </tbody> 
</table>

这个方法主要就是调用redis执行[eval lua][]，为什么使用eval，因为redis对lua脚本执行具有原子性。把这个方法翻译一下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 </td> 
 <td> -- 1. 没被锁{key不存在} eval&nbsp;"return redis.call('exists', KEYS[1])"&nbsp;1 myLock -- (1) 设置Lock为key，uuid:threadId为filed, filed值为1 eval&nbsp;"return redis.call('hset', KEYS[1], ARGV[2], 1)"&nbsp;1 myLock&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- (2) 设置key过期时间{防止获取锁后线程挂掉导致死锁} eval&nbsp;"return redis.call('pexpire', KEYS[1], ARGV[1])"&nbsp;1 myLock&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 &nbsp; -- 2. 已经被同线程获得锁{key存在并且field存在} eval&nbsp;"return redis.call('hexists', KEYS[1], ARGV[2])"&nbsp;1 myLock&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- (1) 可重入，但filed字段+1 eval&nbsp;"return redis.call('hincrby', KEYS[1], ARGV[2],1)"&nbsp;1 myLock&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- (2) 刷新过去时间 eval&nbsp;"return redis.call('pexpire', KEYS[1], ARGV[1])"&nbsp;1 myLock&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 &nbsp; -- 3. 已经被其他线程锁住{key存在，但是field不存在}：以毫秒为单位返回 key 的剩余超时时间 eval&nbsp;"return redis.call('pttl', KEYS[1])"&nbsp;1 myLock </td> 
 </tr> 
 </tbody> 
</table>

这就是核心获取锁的方式，下面直接释放锁方法`unlockInnerAsync`：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 </td> 
 <td> -- 1. key不存在 eval&nbsp;"return redis.call('exists', KEYS[1])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- (1) 发送释放锁的消息，返回1，释放成功 eval&nbsp;"return redis.call('publish', KEYS[2], ARGV[1])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 &nbsp; -- 2. key存在，但field不存在，说明自己不是锁持有者，无权释放，直接return nil eval&nbsp;"return redis.call('hexists', KEYS[1], ARGV[3])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 eval&nbsp;"return nil" &nbsp; -- 3. filed存在，说明是本线程在锁，但有可能其他地方重入锁，不能直接释放，应该-1 eval&nbsp;"return redis.call('hincrby', KEYS[1], ARGV[3]，-1)"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 &nbsp; -- 4. 如果减1后大于0，说明还有其他重入锁，刷新过期时间,返回0。 eval&nbsp;"return redis.call('pexpire', KEYS[1], ARGV[2])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 &nbsp; -- 5. 如果不大于0，说明最后一把锁，需要释放 -- 删除key eval&nbsp;"return redis.call('del', KEYS[1])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- 发释放消息 eval&nbsp;"return redis.call('publish', KEYS[2], ARGV[1])"&nbsp;2 myLock redisson_lock__channel_lock&nbsp;0&nbsp;3000&nbsp;3d7b5418-a86d-48c5-ae15-7fe13ef0034c:110 -- 返回1，释放成功 </td> 
 </tr> 
 </tbody> 
</table>

从释放锁代码中看到，删除key后会发送消息，所以上文提到获取锁失败后，阻塞订阅此消息。

另外，上文提到刷新过期时间方法`scheduleExpirationRenewal`，指线程获取锁后需要不断刷新失效时间，避免未执行完锁就失效。这个方法的实现原理也类似，只是使用了Netty的TimerTask，每到过期时间1/3就去重新刷一次，如果key不存在则停止刷新。Timer实现大概如下：

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 </td> 
 <td> private&nbsp;static&nbsp;void&nbsp;nettyTimer() { final&nbsp;int expireTime =&nbsp;6; EventExecutorGroup group =&nbsp;new DefaultEventExecutorGroup(1); final Timer timer =&nbsp;new HashedWheelTimer(); timer.newTimeout(timerTask -&gt; { Future&lt;Boolean&gt; future = group.submit(() -&gt; { System.out.println("刷新key的失效时间为"+expireTime +"秒"); return&nbsp;false;// 但key不存在时，返回true }); future.addListener(future1 -&gt; { if (!future.getNow()) { nettyTimer(); } }); }, expireTime/3, TimeUnit.SECONDS); } &nbsp; &nbsp; &nbsp; &nbsp; 1. 前因 &nbsp; &nbsp; 以前实现过一个Redis实现的全局锁, 虽然能用, 但是感觉很不完善, 不可重入, 参数太多等等. &nbsp; &nbsp; 最近看到了一个新的Redis客户端Redisson, 看了下源码, 发现了一个比较好的锁实现RLock, 于是记录下. &nbsp; &nbsp; 2. Maven依赖 <pre><code>&lt;dependency&gt;
&nbsp;&nbsp;&nbsp;&nbsp;&lt;groupId&gt;org.redisson&lt;/groupId&gt;
&nbsp;&nbsp;&nbsp;&nbsp;&lt;artifactId&gt;redisson&lt;/artifactId&gt;
&nbsp;&nbsp;&nbsp;&nbsp;&lt;version&gt;1.2.1&lt;/version&gt;
&lt;/dependency&gt;</code></pre> &nbsp; &nbsp; 3. 初试 &nbsp; &nbsp;&nbsp;Redisson中RLock的使用很简单, 来看看一个最简单的例子. <pre><code>import&nbsp;org.redisson.Redisson;
import&nbsp;org.redisson.core.RLock;

public&nbsp;class&nbsp;Temp&nbsp;{

&nbsp;&nbsp;&nbsp;&nbsp;public&nbsp;static&nbsp;void&nbsp;main(String[]&nbsp;args)&nbsp;throws&nbsp;Exception&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Redisson&nbsp;redisson&nbsp;=&nbsp;Redisson.create();

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RLock&nbsp;lock&nbsp;=&nbsp;redisson.getLock("haogrgr");
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lock.lock();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;try&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;System.out.println("hagogrgr");
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;finally&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lock.unlock();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;redisson.shutdown();
&nbsp;&nbsp;&nbsp;&nbsp;}

}</code></pre> &nbsp; &nbsp; 4. RLock接口 &nbsp; &nbsp; 通过上面的例子可以看出, 使用起来和juc里面的Lock接口使用很类似, 那么来看看RLock这个接口. <pre><code>Rlock
|
----------Lock
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------void&nbsp;lock()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------void&nbsp;lockInterruptibly()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------boolean&nbsp;tryLock()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------boolean&nbsp;tryLock(long&nbsp;time,&nbsp;TimeUnit&nbsp;unit)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------void&nbsp;unlock()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------Condition&nbsp;newCondition()
|
----------RObject
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------String&nbsp;getName()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;|
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;----------void&nbsp;delete()
|
----------void&nbsp;lockInterruptibly(long&nbsp;leaseTime,&nbsp;TimeUnit&nbsp;unit)
|
----------boolean&nbsp;tryLock(long&nbsp;waitTime,&nbsp;long&nbsp;leaseTime,&nbsp;TimeUnit&nbsp;unit)
|
----------void&nbsp;lock(long&nbsp;leaseTime,&nbsp;TimeUnit&nbsp;unit)
|
----------void&nbsp;forceUnlock()
|
----------boolean&nbsp;isLocked();
|
----------boolean&nbsp;isHeldByCurrentThread()
|
----------int&nbsp;getHoldCount()</code></pre> &nbsp; &nbsp; 可以看到, 该接口主要继承了Lock接口, 然后扩展了部分方法, 比如:&nbsp; &nbsp; &nbsp; boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) &nbsp; &nbsp; 新加入的leaseTime主要是用来设置锁的过期时间, 形象的解释就是, 如果超过leaseTime还没有解锁的话, 我就强制解锁.&nbsp; &nbsp; &nbsp; 5. RLock接口的实现 &nbsp; &nbsp; 具体的实现类是RedissonLock, 下面来大概看看实现原理. 先看看 (3) 中例子执行时, 所运行的命令(通过monitor命令): <pre><code>127.0.0.1:6379&gt;&nbsp;monitor
OK
1434959509.494805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"SETNX"&nbsp;"haogrgr"&nbsp;"{\"@class\":\"org.redisson.RedissonLock$LockValue\",\"counter\":1,\"id\":\"c374addc-523f-4943-b6e0-c26f7ab061e3\",\"threadId\":1}"
1434959509.494805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"GET"&nbsp;"haogrgr"
1434959509.524805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"MULTI"
1434959509.529805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"DEL"&nbsp;"haogrgr"
1434959509.529805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"PUBLISH"&nbsp;"redisson__lock__channel__{haogrgr}"&nbsp;"0"
1434959509.529805&nbsp;[0&nbsp;127.0.0.1:57911]&nbsp;"EXEC"</code></pre> &nbsp; &nbsp; 可以看到, 大概原理是, 通过判断Redis中是否有某一key, 来判断是加锁还是等待, 最后的publish是一个解锁后, 通知阻塞在lock的线程. &nbsp; &nbsp; &nbsp; 分布式锁的实现依赖的单点, 这里Redis就是单点, 通过在Redis中维护状态信息来实现全局的锁. 那么来看看RedissonLock如何 &nbsp; &nbsp; 实现可重入, 保证原子性等等细节. &nbsp; &nbsp; 6. 加锁源码分析 &nbsp; &nbsp; 从最简单的无参数的lock参数来看源码. <pre><code>public&nbsp;void&nbsp;lock()&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;try&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lockInterruptibly();
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;catch&nbsp;(InterruptedException&nbsp;e)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Thread.currentThread().interrupt();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return;
&nbsp;&nbsp;&nbsp;&nbsp;}
}

public&nbsp;void&nbsp;lockInterruptibly()&nbsp;throws&nbsp;InterruptedException&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;lockInterruptibly(-1,&nbsp;null);&nbsp;&nbsp;&nbsp;&nbsp;//leaseTime&nbsp;:&nbsp;-1&nbsp;表示key不设置过期时间
}

public&nbsp;void&nbsp;lockInterruptibly(long&nbsp;leaseTime,&nbsp;TimeUnit&nbsp;unit)&nbsp;throws&nbsp;InterruptedException&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;Long&nbsp;ttl;
&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(leaseTime&nbsp;!=&nbsp;-1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner(leaseTime,&nbsp;unit);
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner();
&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;lock&nbsp;acquired
&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;==&nbsp;null)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return;
&nbsp;&nbsp;&nbsp;&nbsp;}

&nbsp;&nbsp;&nbsp;&nbsp;subscribe().awaitUninterruptibly();

&nbsp;&nbsp;&nbsp;&nbsp;try&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;(true)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(leaseTime&nbsp;!=&nbsp;-1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner(leaseTime,&nbsp;unit);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;lock&nbsp;acquired
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;==&nbsp;null)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;break;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;waiting&nbsp;for&nbsp;message
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RedissonLockEntry&nbsp;entry&nbsp;=&nbsp;ENTRIES.get(getEntryName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;&gt;=&nbsp;0)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;entry.getLatch().tryAcquire(ttl,&nbsp;TimeUnit.MILLISECONDS);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;entry.getLatch().acquire();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;finally&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;unsubscribe();
&nbsp;&nbsp;&nbsp;&nbsp;}
}</code></pre> &nbsp; &nbsp;&nbsp; &nbsp; &nbsp; 代码有点多, 但是没关系, 慢慢分解, 由于这里我们是调用无参数的lock方法, 所以最后执行到的方法是: <pre><code>private&nbsp;Long&nbsp;tryLockInner()&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;final&nbsp;LockValue&nbsp;currentLock&nbsp;=&nbsp;new&nbsp;LockValue(id,&nbsp;Thread.currentThread().getId());&nbsp;&nbsp;&nbsp;&nbsp;//保存锁的状态:&nbsp;客户端UUID+线程ID来唯一标识某一JVM实例的某一线程
&nbsp;&nbsp;&nbsp;&nbsp;currentLock.incCounter();&nbsp;&nbsp;&nbsp;&nbsp;//用来保存重入次数,&nbsp;实现可重入功能,&nbsp;初始情况是1

&nbsp;&nbsp;&nbsp;&nbsp;//Redisson封装了交互的细节,&nbsp;具体的逻辑为execute方法逻辑.
&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;connectionManager.write(getName(),&nbsp;new&nbsp;SyncOperation&lt;LockValue,&nbsp;Long&gt;()&nbsp;{

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;@Override
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;public&nbsp;Long&nbsp;execute(RedisConnection&lt;Object,&nbsp;LockValue&gt;&nbsp;connection)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Boolean&nbsp;res&nbsp;=&nbsp;connection.setnx(getName(),&nbsp;currentLock);&nbsp;&nbsp;&nbsp;&nbsp;//如果key:haogrgr不存在,&nbsp;就set并返回true,&nbsp;否则返回false
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(!res)&nbsp;{&nbsp;&nbsp;&nbsp;&nbsp;//如果设置失败,&nbsp;那么表示有锁竞争了,&nbsp;于是获取当前锁的状态,&nbsp;如果拥有者是当前线程,&nbsp;就累加重入次数并set新值
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.watch(getName());&nbsp;&nbsp;&nbsp;&nbsp;//通过watch命令配合multi来实现简单的事务功能
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;LockValue&nbsp;lock&nbsp;=&nbsp;(LockValue)&nbsp;connection.get(getName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(lock&nbsp;!=&nbsp;null&nbsp;&amp;&amp;&nbsp;lock.equals(currentLock))&nbsp;{&nbsp;&nbsp;&nbsp;&nbsp;//LockValue的equals实现为比较客户id和threadid是否一样
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lock.incCounter();&nbsp;&nbsp;&nbsp;&nbsp;//如果当前线程已经获取过锁,&nbsp;则累加加锁次数,&nbsp;并set更新
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.multi();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.set(getName(),&nbsp;lock);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(connection.exec().size()&nbsp;==&nbsp;1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;null;&nbsp;&nbsp;&nbsp;&nbsp;//set成功,&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.unwatch();

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//走到这里,&nbsp;说明上面set的时候,&nbsp;其他客户端在&nbsp;&nbsp;watch之后-&gt;set之前&nbsp;有其他客户端修改了key值
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//则获取key的过期时间,&nbsp;如果是永不过期,&nbsp;则返回-1,&nbsp;具体处理后面说明
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;Long&nbsp;ttl&nbsp;=&nbsp;connection.pttl(getName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;ttl;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;null;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;});
}</code></pre> &nbsp; &nbsp; &nbsp; tryLockInner的逻辑已经看完了, &nbsp;可以知道, 有三种情况: &nbsp; &nbsp; (1) key不存在, 加锁:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;当key不存在时, 设置锁的初始状态并set, 具体来看就是 setnx &nbsp; haogrgr &nbsp; LockValue{&nbsp;id:&nbsp;Redisson对象的id, &nbsp;threadId: 当前线程id, &nbsp;counter: 当前重入次数,这里为第一次获取,所以为1} &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;通过上面的操作. 达到获取锁的目的, 通过setnx来达到实现类似于 &nbsp;if(map.get(key) == null) { map.put(key) } 的功能, 防止多个客户端同时set时, 新值覆盖老值. &nbsp; &nbsp; (2)key存在, 且获取锁的当前线程, 重入: &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;这里就是锁重入的情况, 也就是锁的拥有者第二次调用lock方法, 这时, 通过先get, 然后比较客户端ID和当前线程ID来判断拥有锁的线程是不是当前线程.(客户端ID+线程ID才能唯一定位锁拥有者线程) &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;判断发现当前是重入情况, 则累加LockValue的counter, 然后重新set回去, 这里使用到了watch和multi命令, 防止 &nbsp; get -&gt; set &nbsp; 期间其他客户端修改了key的值. &nbsp; &nbsp; (3)key存在, 且是其他线程获取的锁, 等待: &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;首先尝试获取锁(setnx), 失败后发现锁拥有者不是当前线程, 则获取key的过期时间, 返回过期时间 &nbsp; &nbsp; &nbsp; 那么接下来看看tryLockInner调用完成后的处理代码. <pre><code>public&nbsp;void&nbsp;lockInterruptibly(long&nbsp;leaseTime,&nbsp;TimeUnit&nbsp;unit)&nbsp;throws&nbsp;InterruptedException&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;Long&nbsp;ttl;
&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(leaseTime&nbsp;!=&nbsp;-1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner(leaseTime,&nbsp;unit);
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner();&nbsp;&nbsp;&nbsp;//lock()方法调用会走的逻辑
&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;lock&nbsp;acquired
&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;==&nbsp;null)&nbsp;{&nbsp;&nbsp;&nbsp;//加锁成功(新获取锁,&nbsp;重入情况)&nbsp;tryLockInner会返回null,&nbsp;失败会返回key超时时间,&nbsp;或者-1(key未设置超时时间)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return;&nbsp;&nbsp;&nbsp;//加锁成功,&nbsp;返回
&nbsp;&nbsp;&nbsp;&nbsp;}

&nbsp;&nbsp;&nbsp;&nbsp;//subscribe这个方法代码有点多,&nbsp;Redisson通过netty来和redis通讯,&nbsp;然后subscribe返回的是一个Future类型,
&nbsp;&nbsp;&nbsp;&nbsp;//Future的awaitUninterruptibly()调用会阻塞,&nbsp;然后Redisson通过Redis的pubsub来监听unlock的topic(getChannelName())
&nbsp;&nbsp;&nbsp;&nbsp;//例如,&nbsp;5中所看到的命令&nbsp;"PUBLISH"&nbsp;"redisson__lock__channel__{haogrgr}"&nbsp;"0"
&nbsp;&nbsp;&nbsp;&nbsp;//当解锁时,&nbsp;会向名为&nbsp;getChannelName()&nbsp;的topic来发送解锁消息("0")
&nbsp;&nbsp;&nbsp;&nbsp;//而这里&nbsp;subscribe()&nbsp;中监听这个topic,&nbsp;在订阅成功时就会唤醒阻塞在awaitUninterruptibly()的方法.&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;//所以线程在这里只会阻塞很短的时间(订阅成功即唤醒,&nbsp;并不代表已经解锁)
&nbsp;&nbsp;&nbsp;&nbsp;subscribe().awaitUninterruptibly();

&nbsp;&nbsp;&nbsp;&nbsp;try&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;(true)&nbsp;{&nbsp;&nbsp;&nbsp;&nbsp;//循环,&nbsp;不断重试lock
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(leaseTime&nbsp;!=&nbsp;-1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner(leaseTime,&nbsp;unit);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;ttl&nbsp;=&nbsp;tryLockInner();&nbsp;&nbsp;&nbsp;//不多说了
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;lock&nbsp;acquired
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;==&nbsp;null)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;break;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}

&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;这里才是真正的等待解锁消息,&nbsp;收到解锁消息,&nbsp;就唤醒,&nbsp;然后尝试获取锁,&nbsp;成功返回,&nbsp;失败则阻塞在acquire().
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;收到订阅成功消息,&nbsp;则唤醒阻塞上面的subscribe().awaitUninterruptibly();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;收到解锁消息,&nbsp;则唤醒阻塞在下面的entry.getLatch().acquire();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;RedissonLockEntry&nbsp;entry&nbsp;=&nbsp;ENTRIES.get(getEntryName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(ttl&nbsp;&gt;=&nbsp;0)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;entry.getLatch().tryAcquire(ttl,&nbsp;TimeUnit.MILLISECONDS);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;entry.getLatch().acquire();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;finally&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;unsubscribe();&nbsp;&nbsp;//加锁成功或异常,解除订阅
&nbsp;&nbsp;&nbsp;&nbsp;}
}</code></pre> &nbsp; &nbsp; &nbsp; 主要的代码都加上了详细的注释,&nbsp;subscribe() 方法的代码复杂些, 但具体就是利用redis的pubsub提供一个通知机制来减少不断的重试. &nbsp; &nbsp; 很多的Redis锁实现都是失败后sleep一定时间后重试, 在锁被占用时间较长时, 不断的重试是浪费, 而sleep也会导致不必要的时间浪费(在sleep期间可能已经解锁了), sleep时间太长, 时间浪费, 太短, 重试次数会增加~~~. &nbsp; &nbsp; 到这里lock的逻辑已经看完了, 其他的比如tryLock方法逻辑和lock类似, 不过加了超时时间, 然后还有一种lock方法就是对key加上了过期时间. &nbsp; &nbsp; 7. 解锁源码 &nbsp; &nbsp; unlock的逻辑相对简单. <pre><code>public&nbsp;void&nbsp;unlock()&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;connectionManager.write(getName(),&nbsp;new&nbsp;SyncOperation&lt;Object,&nbsp;Void&gt;()&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;@Override
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;public&nbsp;Void&nbsp;execute(RedisConnection&lt;Object,&nbsp;Object&gt;&nbsp;connection)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;LockValue&nbsp;lock&nbsp;=&nbsp;(LockValue)&nbsp;connection.get(getName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(lock&nbsp;!=&nbsp;null)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;LockValue&nbsp;currentLock&nbsp;=&nbsp;new&nbsp;LockValue(id,&nbsp;Thread.currentThread().getId());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(lock.equals(currentLock))&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(lock.getCounter()&nbsp;&gt;&nbsp;1)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;lock.decCounter();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.set(getName(),&nbsp;lock);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;unlock(connection);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;throw&nbsp;new&nbsp;IllegalMonitorStateException("Attempt&nbsp;to&nbsp;unlock&nbsp;lock,&nbsp;not&nbsp;locked&nbsp;by&nbsp;current&nbsp;id:&nbsp;"
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;+&nbsp;id&nbsp;+&nbsp;"&nbsp;thread-id:&nbsp;"&nbsp;+&nbsp;Thread.currentThread().getId());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}&nbsp;else&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;//&nbsp;could&nbsp;be&nbsp;deleted
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return&nbsp;null;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;});
}

private&nbsp;void&nbsp;unlock(RedisConnection&lt;Object,&nbsp;Object&gt;&nbsp;connection)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;int&nbsp;counter&nbsp;=&nbsp;0;
&nbsp;&nbsp;&nbsp;&nbsp;while&nbsp;(counter&nbsp;&lt;&nbsp;5)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.multi();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.del(getName());
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;connection.publish(getChannelName(),&nbsp;unlockMessage);
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;List&lt;Object&gt;&nbsp;res&nbsp;=&nbsp;connection.exec();
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;if&nbsp;(res.size()&nbsp;==&nbsp;2)&nbsp;{
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;return;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;counter++;
&nbsp;&nbsp;&nbsp;&nbsp;}
&nbsp;&nbsp;&nbsp;&nbsp;throw&nbsp;new&nbsp;IllegalStateException("Can't&nbsp;unlock&nbsp;lock&nbsp;after&nbsp;5&nbsp;attempts.&nbsp;Current&nbsp;id:&nbsp;"
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;+&nbsp;id&nbsp;+&nbsp;"&nbsp;thread-id:&nbsp;"&nbsp;+&nbsp;Thread.currentThread().getId());
}</code></pre> &nbsp; &nbsp; &nbsp; 具体的逻辑比较简单, 我就不注释了, 大概就是, 如果是多次重入的, 就以此递减然后 set, 如果是只lock一次的, 就删除, 然后publish一条解锁的message到getChannelName() tocpic. &nbsp; &nbsp; 这里解锁会重试五次, 失败就抛异常. &nbsp; 8.总结 &nbsp; &nbsp; 逻辑并不复杂, 但是通过记录客户端ID和线程ID来唯一标识线程, 实现重入功能, 通过pub sub功能来减少空转. &nbsp; &nbsp; 优点: 实现了Lock的大部分功能, 提供了特殊情况方法(如:强制解锁, 判断当前线程是否已经获取锁, 超时强制解锁等功能), 可重入, 减少重试. &nbsp; &nbsp; 缺点: 使用依赖Redisson, 而Redisson依赖netty, 如果简单使用, 引入了较多的依赖, pub sub的实时性需要测试, 没有监控等功能, 查问题麻烦, 统计功能也没有(例如慢lock日志, 2333333). &nbsp; &nbsp; &nbsp; &nbsp; 参考列表： 
 <ul> 
 <li><a href="https://mp.weixin.qq.com/s?src=3&amp;timestamp=1503125472&amp;ver=1&amp;signature=xqjBIqXRrTSrhO9bVfPMKw*Gg90a6ZTGaG2SA1uH4jONDItU7r1P9K6CaeV56t9GCC5jkjL3Z7ZetOj1sOHMPAt5aOjS*7wCjjwyg2K7Ve9awDBdmyxko5GRCKE*pDWJ6YfqQCu4sCkQ7VbGJWnrKCOdEMQhUh-P4klm627oYUg=" rel="nofollow">一分钟实现分布式锁</a></li> 
 <li><a href="https://github.com/redisson/redisson" rel="nofollow">https://github.com/redisson/redisson</a></li> 
 <li>&nbsp;</li> 
 </ul></td> 
 </tr> 
 </tbody> 
</table>

--------------------

参考列表：

*  [一分钟实现分布式锁][Link 1]
 *  [https://github.com/redisson/redisson][https_github.com_redisson_redisson]

[2017-08-19-16-44-27.jpg]: /images/20220518/46664af28b064e6897716d5f84604c81.png
[Future_Promise]: http://thoreauz.com/2017/08/20/language/java/%E5%9F%BA%E7%A1%80/%E7%BB%99Future%E4%B8%80%E4%B8%AAPromise/
[eval lua]: https://redis.io/commands/eval
[Link 1]: https://mp.weixin.qq.com/s?src=3&timestamp=1503125472&ver=1&signature=xqjBIqXRrTSrhO9bVfPMKw*Gg90a6ZTGaG2SA1uH4jONDItU7r1P9K6CaeV56t9GCC5jkjL3Z7ZetOj1sOHMPAt5aOjS*7wCjjwyg2K7Ve9awDBdmyxko5GRCKE*pDWJ6YfqQCu4sCkQ7VbGJWnrKCOdEMQhUh-P4klm627oYUg=
[https_github.com_redisson_redisson]: https://github.com/redisson/redisson