什么是分布式锁？几种分布式锁分别是怎么实现的？
<h3>推荐学习</h3> <h3><strong><a href="https://links.jianshu.com/go?to=https%3A%2F%2Fwww.bilibili.com%2Fvideo%2FBV19v411P7Cx%2F" target="_blank">分布式-全家桶（面试+技术）：分布式锁+分布式事务+分布式缓存，redis+zk+nginx+mq+kafka等，必须死磕！</a></strong></h3> <div class="image-package"> <div class="image-container"> <div class="image-container-fill"></div> <div class="image-view" data-width="1046" data-height="651"><img data-original-src="//upload-images.jianshu.io/upload_images/24636137-6b36128b7dc16def.png" data-original-width="1046" data-original-height="651" data-original-format="image/png" data-original-filesize="48785" src="https://imgproxy.citrusreader.com/KDEXzrLmfOpZxN8IqL1lQhWRWdIJCaGvS_uD-6AuOSM/raw:1/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yNDYzNjEzNy02YjM2MTI4YjdkYzE2ZGVmLnBuZw" referrerpolicy="no-referrer"></div> </div> <div class="image-caption">image.png</div> </div> <h1>一、什么是分布式锁：</h1> <p><strong>1、什么是分布式锁：</strong></p> <p>分布式锁，即分布式系统中的锁。在单体应用中我们通过锁解决的是<strong>控制共享资源访问</strong>的问题，而分布式锁，就是解决了<strong>分布式系统中控制共享资源访问</strong>的问题。与单体应用不同的是，分布式系统中竞争共享资源的最小<typo id="typo-135" data-origin="粒度" ignoretag="true">粒度</typo>从线程升级成了进程。</p> <p><strong>2、分布式锁应该具备哪些条件：</strong></p> <ul> <li>在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行</li> <li>高可用的获取锁与释放锁</li> <li>高性能的获取锁与释放锁</li> <li>具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）</li> <li>具备锁失效机制，即自动解锁，防止死锁</li> <li>具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败</li> </ul> <p><strong>3、分布式锁的实现方式：</strong></p> <blockquote> <p>基于数据库实现分布式锁基于Zookeeper实现分布式锁基于reids实现分布式锁</p> </blockquote> <p>这篇文章就简单介绍下这几种分布式锁的实现，重点讲解的是基于redis的分布式锁。</p> <h1>二、基于数据库的分布式锁：</h1> <p>基于数据库的锁实现也有两种方式，一是基于数据库表的增删，另一种是基于数据库排他锁。</p> <p><strong>1、基于数据库表的增删：</strong></p> <p>基于数据库表增删是最简单的方式，首先创建一张锁的表主要包含下列字段：类的全路径名+方法名，时间戳等字段。</p> <p>具体的使用方式：当需要锁住某个方法时，往该表中插入一条相关的记录。类的全路径名+方法名是有唯一性约束的，如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。执行完毕之后，需要delete该记录。</p> <p>（这里只是简单介绍一下，对于上述方案可以进行优化，如：应用主从数据库，数据之间双向同步；一旦挂掉快速切换到备库上；做一个定时任务，每隔一定时间把数据库中的超时数据清理一遍；使用while循环，直到insert成功再返回成功；记录当前获得锁的机器的主机信息和线程信息，下次再获取锁的时候先查询数据库，如果当前机器的主机信息和线程信息在数据库可以查到的话，直接把锁分配给他就可以了，实现可重入锁）</p> <p><strong>2、基于数据库排他锁：</strong></p> <p>基于MySql的InnoDB引擎，可以使用以下方法来实现加锁操作：</p> <pre><code>public void lock(){ connection.setAutoCommit(false) int count = 0; while(count < 4){ try{ select * from lock where lock_name=xxx for update; if(结果不为空){ //代表获取到锁 return; } }catch(Exception e){ } //为空或者抛异常的话都表示没有获取到锁 sleep(1000); count++; } throw new LockException(); } </code></pre> <p>在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁。获得排它锁的线程即可获得分布式锁，当获得锁之后，可以执行方法的业务逻辑，执行完方法之后，释放锁connection.commit()。当某条记录被加上排他锁之后，其他线程无法获取排他锁并被阻塞。</p> <p><strong>3、基于数据库锁的优缺点：</strong></p> <p>上面两种方式都是依赖数据库表，一种是通过表中的记录判断当前是否有锁存在，另外一种是通过数据库的排他锁来实现分布式锁。</p> <ul> <li>优点是直接借助数据库，简单容易理解。</li> <li>缺点是操作数据库需要一定的开销，性能问题需要考虑。</li> </ul> <h1>三、基于Zookeeper的分布式锁</h1> <p>基于zookeeper临时有序节点可以实现的分布式锁。每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。 判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。 当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。 （第三方库有 Curator，Curator提供的InterProcessMutex是分布式锁的实现）</p> <p>Zookeeper实现的分布式锁存在两个个缺点：</p> <ul> <li>（1）性能上可能并没有缓存服务那么高，因为每次在创建锁和释放锁的过程中，都要动态创建、销毁瞬时节点来实现锁功能。ZK中创建和删除节点只能通过Leader服务器来执行，然后将数据同步到所有的Follower机器上。</li> <li>（2）zookeeper的并发安全问题：因为可能存在网络抖动，客户端和ZK集群的session连接断了，zk集群以为客户端挂了，就会删除临时节点，这时候其他客户端就可以获取到分布式锁了。</li> </ul> <h1>四、基于redis的分布式锁：</h1> <p>redis命令说明：</p> <p>（1）setnx命令：set if not exists，当且仅当 key 不存在时，将 key 的值设为 value。若给定的 key 已经存在，则 SETNX 不做任何动作。</p> <ul> <li>返回1，说明该进程获得锁，将 key 的值设为 value</li> <li>返回0，说明其他进程已经获得了锁，进程不能进入临界区。</li> </ul> <p>命令格式：setnx lock.key lock.value</p> <p>（2）get命令：获取key的值，如果存在，则返回；如果不存在，则返回nil</p> <p>命令格式：get lock.key</p> <p>（3）getset命令：该方法是原子的，对key设置newValue这个值，并且返回key原来的旧值。</p> <p>命令格式：getset lock.key newValue</p> <p>（4）del命令：删除redis中指定的key</p> <p>命令格式：del lock.key</p> <h1>方案一：基于set命令的分布式锁</h1> <p><strong>1、加锁：使用setnx进行加锁，当该指令返回1时，说明成功获得锁</strong></p> <p><strong>2、解锁：当得到锁的线程执行完任务之后，使用del命令释放锁，以便其他线程可以继续执行setnx命令来获得锁</strong></p> <blockquote> <p>（1）存在的问题：假设线程获取了锁之后，在执行任务的过程中挂掉，来不及显示地执行del命令释放锁，那么竞争该锁的线程都会执行不了，产生死锁的情况。</p> <p>（2）解决方案：设置锁超时时间</p> </blockquote> <p><strong>3、设置锁<typo id="typo-2688" data-origin="超时" ignoretag="true">超时</typo>时间：setnx 的 key 必须设置一个超时时间，以保证即使没有被显式释放，这把锁也要在一定时间后自动释放。可以使用expire命令设置锁超时时间</strong></p> <blockquote> <p>（1）存在问题：</p> <p>setnx 和 expire 不是原子性的操作，假设某个线程执行setnx 命令，成功获得了锁，但是还没来得及执行expire 命令，服务器就挂掉了，这样一来，这把锁就没有设置过期时间了，变成了死锁，别的线程再也没有办法获得锁了。</p> <p>（2）解决方案：redis的set命令支持在获取锁的同时设置key的过期时间</p> </blockquote> <p><strong>4、使用set命令加锁并设置锁过期时间：</strong></p> <p>命令格式：set <lock.key> <lock.value> nx ex <expireTime></p> <p>详情参考redis使用文档：<a href="https://links.jianshu.com/go?to=http%3A%2F%2Fdoc.redisfans.com%2Fstring%2Fset.html" target="_blank">http://doc.redisfans.com/string/set.html</a></p> <p><strong>（1）存在问题：</strong></p> <p>① 假如线程A成功得到了锁，并且设置的超时时间是 30 秒。如果某些原因导致线程 A 执行<typo id="typo-3121" data-origin="的" ignoretag="true">的</typo>很慢，过了 30 秒都没执行完，这时候锁过期自动释放，线程 B 得到了锁。</p> <p>② 随后，线程A执行完任务，接着执行del指令来释放锁。但这时候线程 B 还没执行完，线程A实际上删除的是线程B加的锁。</p> <p><strong>（2）解决方案：</strong></p> <p>可以在 del 释放锁之前做一个判断，验证当前的锁是不是自己加的锁。在加锁的时候把当前的线程 ID 当做value，并在删除之前验证 key 对应的 value 是不是自己线程的 ID。但是，这样做其实隐含了一个新的问题，get操作、判断和释放锁是两个独立操作，不是原子性。对于非原子性的问题，我们可以使用Lua脚本来确保操作的原子性</p> <p><strong>5、锁续期：</strong>（这种机制类似于redisson的看门狗机制，文章后面会详细说明）</p> <p>虽然步骤4避免了线程A误删掉key的情况，但是同一时间有 A，B 两个线程在访问代码块，仍然是不完美的。怎么办呢？我们可以让获得锁的线程开启一个<strong>守护线程</strong>，用来给快要过期的锁“续期”。</p> <blockquote> <p>① 假设线程A执行了29 秒后还没执行完，这时候守护线程会执行 expire 指令，为这把锁续期 20 秒。守护线程从第 29 秒开始执行，每 20 秒执行一次。</p> <p>② 情况一：当线程A执行完任务，会显式关掉守护线程。</p> <p>③ 情况二：如果服务器忽然断电，由于线程 A 和守护线程在同一个进程，守护线程也会停下。这把锁到了超时的时候，没人给它续命，也就自动释放了。</p> </blockquote> <h1>方案二：基于setnx、get、getset的分布式锁</h1> <p><strong>1、实现原理：</strong></p> <p>（1）setnx(lockkey, 当前时间+过期超时时间) ，如果返回1，则获取锁成功；如果返回0则没有获取到锁，转向步骤(2)</p> <p>（2）get(lockkey)获取值oldExpireTime ，并将这个value值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向步骤(3)</p> <p>（3）计算新的过期时间 newExpireTime=当前时间+锁超时时间，然后getset(lockkey, newExpireTime) 会返回当前lockkey的值currentExpireTime</p> <p>（4）判断 currentExpireTime 与 oldExpireTime 是否相等，如果相等，说明当前getset设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。</p> <p>（5）在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行del命令释放锁（释放锁之前需要判断持有锁的线程是不是当前线程）；如果大于锁设置的超时时间，则不需要再锁进行处理。</p> <p><strong>2、代码实现：</strong></p> <p>（1）获取锁的实现方式：</p> <pre><code>public boolean lock(long acquireTimeout, TimeUnit timeUnit) throws InterruptedException { acquireTimeout = timeUnit.toMillis(acquireTimeout); long acquireTime = acquireTimeout + System.currentTimeMillis(); //使用J.U.C的ReentrantLock threadLock.tryLock(acquireTimeout, timeUnit); try { //循环尝试 while (true) { //调用tryLock boolean hasLock = tryLock(); if (hasLock) { //获取锁成功 return true; } else if (acquireTime < System.currentTimeMillis()) { break; } Thread.sleep(sleepTime); } } finally { if (threadLock.isHeldByCurrentThread()) { threadLock.unlock(); } } return false; } public boolean tryLock() { long currentTime = System.currentTimeMillis(); String expires = String.valueOf(timeout + currentTime); //设置互斥量 if (redisHelper.setNx(mutex, expires) > 0) { //获取锁，设置超时时间 setLockStatus(expires); return true; } else { String currentLockTime = redisUtil.get(mutex); //检查锁是否超时 if (Objects.nonNull(currentLockTime) && Long.parseLong(currentLockTime) < currentTime) { //获取旧的锁时间并设置互斥量 String oldLockTime = redisHelper.getSet(mutex, expires); //旧值与当前时间比较 if (Objects.nonNull(oldLockTime) && Objects.equals(oldLockTime, currentLockTime)) { //获取锁，设置超时时间 setLockStatus(expires); return true; } } return false; } } </code></pre> <p>tryLock方法中，主要逻辑如下：lock调用tryLock方法，参数为获取的超时时间与单位，线程在超时时间内，获取锁操作将自旋在那里，直到该自旋锁的保持者释放了锁。</p> <p>（2）释放锁的实现方式：</p> <pre><code>public boolean unlock() { //只有锁的持有线程才能解锁 if (lockHolder == Thread.currentThread()) { //判断锁是否超时，没有超时才将互斥量删除 if (lockExpiresTime > System.currentTimeMillis()) { redisHelper.del(mutex); logger.info("删除互斥量[{}]", mutex); } lockHolder = null; logger.info("释放[{}]锁成功", mutex); return true; } else { throw new IllegalMonitorStateException("没有获取到锁的线程无法执行解锁操作"); } } </code></pre> <p><strong>存在问题：</strong></p> <p>（1）这个锁的核心是基于System.currentTimeMillis()，如果多台服务器时间不一致，那么问题就出现了，但是这个bug完全可以从服务器运维层面规避的，而且如果服务器时间不一样的话，只要和时间相关的逻辑都是会出问题的</p> <p>（2）如果前一个锁超时的时候，刚好有多台服务器去请求获取锁，那么就会出现同时执行redis.getset()而导致出现过期时间覆盖问题，不过这种情况并不会对正确结果造成影响</p> <p>（3）存在多个线程同时持有锁的情况：如果线程A执行任务的时间超过锁的过期时间，这时另一个线程就可以获得这个锁了，造成多个线程同时持有锁的情况。类似于方案一，可以使用“锁续期”的方式来解决。</p> <p><strong>前两种redis分布式锁的存在的问题</strong></p> <p>前面两种redis分布式锁的实现方式，如果从“高可用”的层面来看，仍然是有所欠缺，也就是说当 redis 是单点的情况下，当发生故障时，则整个业务的分布式锁都将无法使用。</p> <p>为了提高可用性，我们可以使用主从模式或者哨兵模式，但在这种情况下仍然存在问题，在主从模式或者哨兵模式下，正常情况下，如果加锁成功了，那么master节点会异步复制给对应的slave节点。但是如果在这个过程中发生master节点宕机，主备切换，slave节点从变为了 master节点，而锁还没从旧master节点同步过来，这就发生了锁丢失，会导致多个客户端可以同时持有同一把锁的问题。来看个图来想下这个过程：</p> <div class="image-package"> <div class="image-container"> <div class="image-container-fill"></div> <div class="image-view" data-width="380" data-height="420"><img data-original-src="//upload-images.jianshu.io/upload_images/24636137-bdef0f1af6af6168.png" data-original-width="380" data-original-height="420" data-original-format="image/png" data-original-filesize="39665" src="https://imgproxy.citrusreader.com/0ei_0hTG9ucgC6fvxN9CjzODavLCGc0uKVp_ksyNRWU/raw:1/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yNDYzNjEzNy1iZGVmMGYxYWY2YWY2MTY4LnBuZw" referrerpolicy="no-referrer"></div> </div> <div class="image-caption">image.png</div> </div> <p>那么，如何避免这种情况呢？redis 官方给出了基于多个 redis 集群部署的高可用分布式锁解决方案：RedLock，在方案三我们就来详细介绍一下。（备注：如果master节点宕机期间，可以容忍多个客户端同时持有锁，那么就不需要redLock）</p> <h1>方案三：基于RedLock的分布式锁</h1> <p>redLock的官方文档地址：<a href="https://links.jianshu.com/go?to=https%3A%2F%2Fredis.io%2Ftopics%2Fdistlock" target="_blank">https://redis.io/topics/distlock</a></p> <p>Redlock算法是Redis的作者 Antirez 在单Redis节点基础上引入的高可用模式。Redlock的加锁要结合单节点分布式锁算法共同实现，因为它是RedLock的基础</p> <p><strong>1、加锁实现原理：</strong></p> <p>现在假设有5个Redis主节点(大于3的奇数个)，这样基本保证他们不会同时都宕掉，获取锁和释放锁的过程中，客户端会执行以下操作：</p> <p>（1）获取当前Unix时间，以毫秒为单位，并设置超时时间TTL</p> <blockquote> <p>TTL 要大于 正常业务执行的时间 + 获取所有redis服务消耗时间 + 时钟漂移</p> </blockquote> <p>（2）依次尝试从5个实例，使用相同的key和具有唯一性的value获取锁，当向Redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间TTL，这样可以避免客户端死等。比如：TTL为5s，设置获取锁最多用1s，所以如果一秒内无法获取锁，就放弃获取这个锁，从而尝试获取下个锁</p> <p>（3）客户端 获取所有能获取的锁后的时间 减去 第(1)步的时间，就得到锁的获取时间。锁的获取时间要小于锁失效时间TTL，并且至少从半数以上的Redis节点取到锁，才算获取成功锁</p> <p>（4）如果成功获得锁，key的真正有效时间 = TTL - 锁的获取时间 - 时钟漂移。比如：TTL 是5s,获取所有锁用了2s，则真正锁有效时间为3s</p> <p>（5）如果因为某些原因，获取锁失败（没有在半数以上实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁，无论Redis实例是否加锁成功，因为可能服务端响应消息丢失了但是实际成功了。</p> <blockquote> <p>设想这样一种情况：客户端发给某个Redis节点的获取锁的请求成功到达了该Redis节点，这个节点也成功执行了SET操作，但是它返回给客户端的响应包却丢失了。这在客户端看来，获取锁的请求由于超时而失败了，但在Redis这边看来，加锁已经成功了。因此，释放锁的时候，客户端也应该对当时获取锁失败的那些Redis节点同样发起请求。实际上，这种情况在异步通信模型中是有可能发生的：客户端向服务器通信是正常的，但反方向却是有问题的。</p> </blockquote> <p>（6）失败重试：当client不能获取锁时，应该在随机时间后重试获取锁；同时重试获取锁要有一定次数限制；</p> <blockquote> <p>在随机时间后进行重试，主要是防止过多的客户端同时尝试去获取锁，导致彼此都获取锁失败的问题。</p> </blockquote> <p>算法示意图如下：</p> <div class="image-package"> <div class="image-container"> <div class="image-container-fill"></div> <div class="image-view" data-width="581" data-height="580"><img data-original-src="//upload-images.jianshu.io/upload_images/24636137-8faa3f124b379f4e.png" data-original-width="581" data-original-height="580" data-original-format="image/png" data-original-filesize="70501" src="https://imgproxy.citrusreader.com/V5h5DicShz_6_66aEBTjHvBZ5nBCcYfoAzV7bp2XEmQ/raw:1/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yNDYzNjEzNy04ZmFhM2YxMjRiMzc5ZjRlLnBuZw" referrerpolicy="no-referrer"></div> </div> <div class="image-caption">image.png</div> </div> <p><strong>2、RedLock性能及崩溃恢复的相关解决方法：</strong></p> <p>由于N个Redis节点中的大多数能正常工作就能保证Redlock正常工作，因此理论上它的可用性更高。前面我们说的主从架构下存在的安全性问题，在RedLock中已经不存在了，但如果有节点发生崩溃重启，还是会对锁的安全性有影响的，具体的影响程度跟Redis持久化配置有关：</p> <p>（1）如果redis没有持久化功能，在clientA获取锁成功后，所有redis重启，clientB能够再次获取到锁，这样违法了锁的排他互斥性；</p> <p>（2）如果启动AOF永久化存储，事情会好些， 举例：当我们重启redis后，由于redis过期机制是按照unix时间戳走的，所以在重启后，然后会按照规定的时间过期，不影响业务；但是由于AOF同步到磁盘的方式默认是每秒一次，如果在一秒内断电，会导致数据丢失，立即重启会造成锁互斥性失效；但如果同步磁盘方式使用Always(每一个写命令都同步到硬盘)造成性能急剧下降；所以在锁完全有效性和性能方面要有所取舍；</p> <p>（3）为了有效解决既保证锁完全有效性 和 性能高效问题：antirez又提出了“延迟重启”的概念，redis同步到磁盘方式保持默认的每秒1次，在redis崩溃单机后（无论是一个还是所有），先不立即重启它，而是等待TTL时间后再重启，这样的话，这个节点在重启前所参与的锁都会过期，它在重启后就不会对现有的锁造成影响，缺点是在TTL时间内服务相当于暂停状态；</p> <p><strong>3、Redisson中RedLock的实现：</strong></p> <p>在JAVA的redisson包已经实现了对RedLock的封装，主要是通过 redisClient 与 lua 脚本实现的，之所以使用 lua 脚本，是为了实现加解锁校验与执行的事务性。</p> <p><strong>（1）唯一ID的生成：</strong></p> <p>分布式事务锁中，为了能够让作为中心节点的存储节点获取锁的持有者，从而避免锁被非持有者误解锁，每个发起请求的 client 节点都必须具有全局唯一的 id。通常我们是使用 UUID 来作为这个唯一 id，redisson 也是这样实现的，在此基础上，redisson 还加入了 threadid 避免了多个线程反复获取 UUID 的性能损耗</p> <pre><code>protected final UUID id = UUID.randomUUID(); String getLockName(long threadId) { return id + ":" + threadId; } </code></pre> <p><strong>（2）加锁逻辑：</strong></p> <p>redisson 加锁的核心代码非常容易理解，通过传入 TTL 与唯一 id，实现一段时间的加锁请求。下面是可重入锁的实现逻辑：</p> <pre><code><T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { internalLockLeaseTime = unit.toMillis(leaseTime); // 获取锁时向5个redis实例发送的命令 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, // 校验分布式锁的KEY是否已存在，如果不存在，那么执行hset命令（hset REDLOCK_KEY uuid+threadId 1），并通过pexpire设置失效时间（也是锁的租约时间） "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; " + // 如果分布式锁的KEY已存在，则校验唯一 id，如果唯一 id 匹配，表示是当前线程持有的锁，那么重入次数加1，并且设置失效时间 "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; " + // 获取分布式锁的KEY的失效时间毫秒数 "return redis.call('pttl', KEYS[1]);", // KEYS[1] 对应分布式锁的 key；ARGV[1] 对应 TTL；ARGV[2] 对应唯一 id Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); } </code></pre> <p><strong>（3）释放锁逻辑：</strong></p> <pre><code>protected RFuture<Boolean> unlockInnerAsync(long threadId) { // 向5个redis实例都执行如下命令 return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, // 如果分布式锁 KEY 不存在，那么向 channel 发布一条消息 "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end;" + // 如果分布式锁存在，但是唯一 id 不匹配，表示锁已经被占用 "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + // 如果就是当前线程占有分布式锁，那么将重入次数减 1 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + // 重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只设置失效时间，不删除 "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "else " + // 重入次数减1后的值如果为0，则删除锁，并发布解锁消息 "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", // KEYS[1] 表示锁的 key，KEYS[2] 表示 channel name，ARGV[1] 表示解锁消息，ARGV[2] 表示 TTL，ARGV[3] 表示唯一 id Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId)); } </code></pre> <p><strong>（4）redisson中RedLock的使用：</strong></p> <pre><code>Config config = new Config(); config.useSentinelServers() .addSentinelAddress("127.0.0.1:6369","127.0.0.1:6379", "127.0.0.1:6389") .setMasterName("masterName") .setPassword("password").setDatabase(0); RedissonClient redissonClient = Redisson.create(config); RLock redLock = redissonClient.getLock("REDLOCK_KEY"); try { // 尝试加锁，最多等待500ms，上锁以后10s自动解锁 boolean isLock = redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS); if (isLock) { //获取锁成功，执行对应的业务逻辑 } } catch (Exception e) { e.printStackTrace(); } finally { redLock.unlock(); } </code></pre> <p>可以看到，redisson 包的实现中，通过 lua 脚本校验了解锁时的 client 身份，所以我们无需再在 finally 中去判断是否加锁成功，也无需做额外的身份校验，可以说已经达到开箱即用的程度了。</p> <p>同样，基于RedLock实现的分布式锁也存在 client 获取锁之后，在 TTL 时间内没有完成业务逻辑的处理，而此时锁会被自动释放，造成多个线程同时持有锁的问题。而Redisson 在实现的过程中，自然也考虑到了这一问题，redisson 提供了一个“看门狗”的特性，当锁即将过期还没有释放时，不断的延长锁key的生存时间。（具体实现原理会在方案四进行介绍）</p> <h1>方案四：基于Redisson看门狗的分布式锁</h1> <p>前面说了，如果某些原因导致持有锁的线程在锁过期时间内，还没执行完任务，而锁因为还没超时被自动释放了，那么就会导致多个线程同时持有锁的现象出现，而为了解决这个问题，可以进行“锁续期”。其实，在JAVA的Redisson包中有一个"看门狗"机制，已经帮我们实现了这个功能。</p> <p><strong>1、redisson原理：</strong></p> <p>redisson在获取锁之后，会维护一个看门狗线程，当锁即将过期还没有释放时，不断的延长锁key的生存时间</p> <div class="image-package"> <div class="image-container"> <div class="image-container-fill"></div> <div class="image-view" data-width="1035" data-height="323"><img data-original-src="//upload-images.jianshu.io/upload_images/24636137-af444596f5e7db12.png" data-original-width="1035" data-original-height="323" data-original-format="image/png" data-original-filesize="50473" src="https://imgproxy.citrusreader.com/zqFEZFADJCTj_Jk-476v2Ulz-0Rn_ng-ZCtjB6qZB_s/raw:1/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yNDYzNjEzNy1hZjQ0NDU5NmY1ZTdkYjEyLnBuZw" referrerpolicy="no-referrer"></div> </div> <div class="image-caption">image.png</div> </div> <p><strong>2、加锁机制：</strong></p> <p>线程去获取锁，获取成功：执行lua脚本，保存数据到redis数据库。</p> <p>线程去获取锁，获取失败：一直通过while循环尝试获取锁，获取成功后，执行lua脚本，保存数据到redis数据库。</p> <p><strong>3、watch dog自动延期机制：</strong></p> <blockquote> <p>看门狗启动后，对整体性能也会有一定影响，默认情况下看门狗线程是不启动的。如果使用redisson进行加锁的同时设置了锁的过期时间，也会导致看门狗机制失效。</p> </blockquote> <p>redisson在获取锁之后，会维护一个看门狗线程，在每一个锁设置的过期时间的1/3处，如果线程还没执行完任务，则不断延长锁的有效期。看门狗的检查锁超时时间默认是30秒，可以通过 lockWactchdogTimeout 参数来改变。</p> <p>加锁的时间默认是30秒，如果加锁的业务没有执行完，那么每隔 30 ÷ 3 = 10秒，就会进行一次续期，把锁重置成30秒，保证解锁前锁不会自动失效。</p> <p>那万一业务的机器宕机了呢？如果宕机了，那看门狗线程就执行不了了，就续不了期，那自然30秒之后锁就解开了呗。</p> <p><strong>4、redisson分布式锁的关键点：</strong></p> <blockquote> <p>a. 对key不设置过期时间，由Redisson在加锁成功后给维护一个watchdog看门狗，watchdog负责定时监听并处理，在锁没有被释放且快要过期的时候自动对锁进行续期，保证解锁前锁不会自动失效</p> <p>b. 通过Lua脚本实现了加锁和解锁的原子操作</p> <p>c. 通过记录获取锁的客户端id，每次加锁时判断是否是当前客户端已经获得锁，实现了可重入锁。</p> </blockquote> <p><strong>5、Redisson的使用：</strong></p> <p>在方案三中，我们已经演示了基于Redisson的RedLock的使用案例，其实 Redisson 也封装 可重入锁（Reentrant Lock）、公平锁（Fair Lock）、联锁（MultiLock）、红锁（RedLock）、读写锁（ReadWriteLock）、 信号量（Semaphore）、可过期性信号量（PermitExpirableSemaphore）、 闭锁（CountDownLatch）等，具体使用说明可以参考官方文档：Redisson的分布式锁和同步器</p> <p>附：redLock的官方文档翻译</p> <div class="image-package"> <div class="image-container"> <div class="image-container-fill"></div> <div class="image-view" data-width="1430" data-height="6670"><img data-original-src="//upload-images.jianshu.io/upload_images/24636137-c101a541ee090a8d.png" data-original-width="1430" data-original-height="6670" data-original-format="image/png" data-original-filesize="1686035" src="https://imgproxy.citrusreader.com/g1hsH45OpM8LI3vCaxnwzLG-E3rQ4NWXPhoVpJqJERg/raw:1/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yNDYzNjEzNy1jMTAxYTU0MWVlMDkwYThkLnBuZw" referrerpolicy="no-referrer"></div> </div> <div class="image-caption">image.png</div> </div> <blockquote> <p>作者：张维鹏<br> 原文链接：<a href="https://links.jianshu.com/go?to=https%3A%2F%2Fblog.csdn.net%2Fa745233700%2Farticle%2Fdetails%2F88084219" target="_blank">https://blog.csdn.net/a745233700/article/details/88084219</a></p> </blockquote>