作者:京东保险 张江涛
1、为什么要有分布式锁?
JUC提供的锁机制,能够保障在同一个JVM过程中同一时刻只有一个线程执行操作逻辑;
多服务多节点的状况下,就意味着有多个JVM过程,要做到这样,就须要有一个中间人;
分布式锁就是用来保障在同一时刻,仅有一个JVM过程中的一个线程在执行操作逻辑;
换句话说,JUC的锁和分布式锁都是一种爱护系统资源的措施。尽可能将并发带来的不确定性转换为同步的确定性;
2、分布式锁个性(五大个性 十分重要)
个性1:互斥性。在任意时刻,只有一个客户端能持有锁。
个性2: 不会产生死锁。即便有一个客户端在持有锁的期间解体而没有被动解锁,也能保障后续其余客户端能加锁。
个性3: 解铃还须系铃人。加锁和解锁必须是同一个客户端(线程),客户端本人不能把他人加的锁给解了。
个性4:可重入性。同一个现线程曾经获取到锁,可再次获取到锁。
个性5: 具备容错性。只有大部分的分布式锁节点失常运行,客户端就能够加锁和解锁。
2-1 常见分布式锁的三种实现形式
1. 数据库锁;2. 基于ZooKeeper的分布式锁;3. 基于Redis的分布式锁。
2-2 本文咱们次要聊 redis实现分布式锁:
一个 setnx 就行了?value没意义?还有人认为 incr 也能够?再加个超时工夫就行了?
3、分布式锁个性2之不会产生死锁
很多线程去上锁,谁锁胜利谁就有权力执行操作逻辑,其余线程要么间接走抢锁失败的逻辑,要么自旋尝试抢锁;
• 比方说 A线程竞争到了锁,开始执行操作逻辑(代码逻辑演示中,应用 Jedis客户端为例);
public static void doSomething() { // RedisLock是封装好的一个类 RedisLock redisLock = new RedisLock(jedis); // 创立jedis实例的代码省略,不是重点 try { redisLock.lock(); // 上锁 // 解决业务 System.out.println(Thread.currentThread().getName() + " 线程解决业务逻辑中..."); Thread.sleep(2000); System.out.println(Thread.currentThread().getName() + " 线程解决业务逻辑结束"); redisLock.unlock(); // 开释锁 } catch (Exception e) { e.printStackTrace(); }}
• 失常状况下,A 线程执行完操作逻辑后,应该将锁开释。如果说执行过程中抛出异样,程序不再持续走失常的开释锁流程,没有开释锁怎么办?所以咱们想到:
• 开释锁的流程肯定要在 finally{} 块中执行,当然,上锁的流程肯定要在 finally{} 对应的 try{} 块中,否则 finally{} 就没用了,如下:
public static void doSomething() { RedisLock redisLock = new RedisLock(jedis); // 创立jedis实例的代码省略,不是重点 try { redisLock.lock(); // 上锁,必须在 try{}中 // 解决业务 System.out.println(Thread.currentThread().getName() + " 线程解决业务逻辑中..."); Thread.sleep(2000); System.out.println(Thread.currentThread().getName() + " 线程解决业务逻辑结束"); } catch (Exception e) { e.printStackTrace(); } finally { redisLock.unlock(); // 在finally{} 中开释锁 }}
写法留神: redisLock.lock(); 上分布式锁,必须在 try{}中。
在JAVA多线程中 lock.lock(); 单机多线程加锁操作须要在try{}之前。
3-1 redisLock.unlock() 放在 finally{} 块中就行了吗?还须要设置超时工夫
如果在执行 try{} 中逻辑的时候,程序呈现了 System.exit(0); 或者 finally{} 中执行异样,比方说连贯不上 redis-server了;或者还未执行到 finally{}的时候,JVM过程挂掉了,服务宕机;这些状况都会导致没有胜利开释锁,别的线程始终拿不到锁,怎么办?如果我的零碎因为一个节点影响,别的节点也都无奈失常提供服务了,那我的零碎也太弱了。所以咱们想到必须要将危险升高,能够给锁设置一个超时工夫,比方说 1秒,即使产生了上边的状况,那我的锁也会在 1秒之后主动开释,其余线程就能够获取到锁,接班干活了;
public static final String lock_key = "zjt-lock"; public void lock() { while (!tryLock()) { try { Thread.sleep(50); // 在while中自旋,如果说读者想设置一些自旋次数,期待最大时长等本人去扩大,不是此处的重点 } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println("线程:" + threadName + ",占锁胜利!★★★"); } private boolean tryLock() { SetParams setParams = new SetParams(); setParams.ex(1); // 超时工夫1s setParams.nx(); // nx String response = jedis.set(lock_key, "", setParams); // 转换为redis命令就是:set zjt-key "" ex 1 nx return "OK".equals(response); }
留神,上锁的时候,设置key和设置超时工夫这两个操作要是原子性的,要么都执行,要么都不执行。
Redis原生反对:
// http://redis.io/commands/set.htmlSET key value [EX seconds] [PX milliseconds] [NX|XX]
不要在代码里边分两次调用:
set k vexipre k time
3-2 锁的超时工夫该怎么计算?
方才假如的超时工夫 1s是怎么计算的?这个工夫该设多少适合呢?
锁中的业务逻辑的执行工夫,个别是咱们在测试环境进行屡次测试,而后在压测环境多轮压测之后,比方说计算出均匀的执行工夫是 200ms,锁的超时工夫放大3-5倍,比方这里咱们设置为 1s,为啥要放大,因为如果锁的操作逻辑中有网络 IO操作,线上的网络不会总一帆风顺,咱们要给网络抖动留有缓冲工夫。另外,如果你设置 10s,果然产生了宕机,那意味着这 10s两头,你的这个分布式锁的服务全副节点都是不可用的,这个和你的业务以及零碎的可用性有挂钩,要掂量,要谨慎(后边3-13会再具体聊)。那如果一个节点宕机之后能够告诉 redis-server开释锁吗?留神,我是宕机,不可控力,断电了兄弟,告诉不了的。
回头一想,如果我是优雅停机呢,我不是 kill -9,也不是断电,这样仿佛能够去做一些编码去开释锁,你能够参考下 JVM的钩子、Dubbo的优雅停机、或者 linux过程级通信技术来做这件事件。当然也能够手动停服务后,手动删除掉 redis中的锁。
4、分布式锁个性3:解铃还须系铃人
如果说 A线程在执行操作逻辑的过程中,别的线程间接进行了开释锁的操作,是不是就出问题了?
什么?别的线程没有取得锁却间接执行了开释锁??当初是 A线程上的锁,那必定只能 A线程开释锁呀!别的线程开释锁算怎么回事?联想 ReentrantLock中的 isHeldByCurrentThread()办法,所以咱们想到,必须在锁上加个标记,只有上锁的线程 A线程晓得,相当于是一个密语,也就是说开释锁的时候,首先先把密语和锁上的标记进行匹配,如果匹配不上,就没有权力开释锁;
private boolean tryLock() { SetParams setParams = new SetParams(); setParams.ex(1); // 超时工夫1s setParams.nx(); // nx String response = jedis.set(lock_key, "", setParams); // 转换为redis命令就是:set zjt_key "" ex 1 nx return "OK".equals(response); } // 别的线程间接调用开释锁操作,分布式锁解体! public void unlock() { jedis.del(encode(lock_key)); System.out.println("线程:" + threadName + " 开释锁胜利!☆☆☆"); } private byte[] encode(String param) { return param.getBytes(); }
4-1 这个密语value(约定)设置成什么呢?
很多同学说设置成一个 UUID就行了,上锁之前,在该线程代码中生成一个 UUID,将这个作为秘钥,存在锁键的 value中,开释锁的时候,用这个进行校验,因为只有上锁的线程晓得这个秘钥,别的线程是不晓得的。这个可行吗,当然可行。
String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + "then\n" + " return redis.call(\"del\", KEYS[1])\n" + "else\n" + " return 0\n" + "end"; private boolean tryLock(String uuid) { SetParams setParams = new SetParams(); setParams.ex(1); // 超时工夫1s setParams.nx(); // nx String response = jedis.set(lock_key, uuid, setParams); // 转换为redis命令就是:set zjt-key "" ex 1 nx return "OK".equals(response); } public void unlock(String uuid) { List<byte[]> keys = Arrays.asList(encode(lock_key)); List<byte[]> args = Arrays.asList(encode(uuid)); // 应用lua脚本,保障原子性 long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args); if (eval == 1) { System.out.println("线程:" + threadName + " 开释锁胜利!☆☆☆"); } else { System.out.println("线程:" + threadName + " 开释锁失败!该线程未持有锁!!!"); } } private byte[] encode(String param) { return param.getBytes(); }
为什么应用 lua脚本?因为保障原子性
因为是两个操作,如果分两步那就是:
get k // 进行秘钥 value的比对del k // 比对胜利后,删除k
如果第一步比对胜利后,第二步还没来得及执行的时候,锁到期,而后紧接着别的线程获取到锁,里边的 uuid曾经变了,也就是说持有锁的线程曾经不是该线程了,此时再执行第二步的删除锁操作,必定是谬误的了。
5.分布式锁个性4之可重入性
作为一把锁,咱们在应用 synchronized、ReentrantLock的时候是不是有可重入性?
那咱们这把分布式锁该如何实现可重入呢?如果 A线程的锁办法逻辑中调用了 x()办法,x()办法中也须要获取这把锁,依照这个逻辑,x()办法中的锁应该重入进去即可,那是不是须要将方才生成的这个 UUID秘钥传递给 x()办法?怎么传递?用参数传递就会侵入业务代码
5-1 不侵入业务代码实现可重入:Thread-Id
咱们次要是想给上锁的 A线程设置一个只有它本人晓得的秘钥,把思路时钟往回拨,想想:
线程自身的 id(Thread.currentThread().getId())是不是就是一个惟一标识呢?咱们把秘钥 value设置为线程的 id不就行了。
String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + "then\n" + " return redis.call(\"del\", KEYS[1])\n" + "else\n" + " return 0\n" + "end"; String addLockLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + "then\n" + " return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + "else\n" + " return 0\n" + "end"; public void lock() { // 判断是否可重入 if (isHeldByCurrentThread()) { return; } while (!tryLock()) { try { Thread.sleep(50); // 自旋 } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println("线程:" + threadName + ",占锁胜利!★★★"); } // 是否是以后线程占有锁,同时将超时工夫从新设置,这个很重要,同样也是原子操作 private boolean isHeldByCurrentThread() { List<byte[]> keys = Arrays.asList(encode(lock_key)); List<byte[]> args = Arrays.asList(encode(String.valueOf(threadId)), encode(String.valueOf(1))); long eval = (Long) jedis.eval(encode(addLockLife_lua), keys, args); return eval == 1; } private boolean tryLock(String uuid) { SetParams setParams = new SetParams(); setParams.ex(1); // 超时工夫1s setParams.nx(); // nx String response = jedis.set(lock_key, String.valueOf(threadId), setParams); // 转换为redis命令就是:set zjt-key xxx ex 1 nx return "OK".equals(response); } public void unlock(String uuid) { List<byte[]> keys = Arrays.asList(encode(lock_key)); List<byte[]> args = Arrays.asList(encode(String.valueOf(threadId))); // 应用lua脚本,保障原子性 long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args); if (eval == 1) { System.out.println("线程:" + threadName + " 开释锁胜利!☆☆☆"); } else { System.out.println("线程:" + threadName + " 开释锁失败!该线程未持有锁!!!"); } } private byte[] encode(String param) { return param.getBytes(); }
5-2 Thread-Id 真能行吗?不行。
想想,咱们说一个 Thread的id是惟一的,是在同一个 JVM过程中,是在一个操作系统中,也就是在一个机器中。而事实是,咱们的部署是集群部署,多个实例节点,那意味着会存在这样一种状况,S1机器上的线程上锁胜利,此时锁中秘钥 value是线程id=1,如果说同一时间 S2机器中,正好线程id=1的线程尝试取得这把锁,比对秘钥发现胜利,后果也重入了这把锁,也开始执行逻辑,此时,咱们的分布式锁解体!怎么解决?咱们只须要在每个节点中保护不同的标识即可,怎么保护呢?利用启动的时候,应用 UUID生成一个惟一标识 APP_ID,放在内存中(或者应用zookeeper去调配机器id等等)。此时,咱们的秘钥 value这样存即可:APP_ID+ThreadId
// static变量,final润饰,加载在内存中,JVM过程生命周期中不变 private static final String APP_ID = UUID.randomUUID().toString(); String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + "then\n" + " return redis.call(\"del\", KEYS[1])\n" + "else\n" + " return 0\n" + "end"; String addLockLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + "then\n" + " return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + "else\n" + " return 0\n" + "end"; public void lock() { // 判断是否可重入 if (isHeldByCurrentThread()) { return; } while (!tryLock()) { try { Thread.sleep(50); // 自旋 } catch (InterruptedException e) { e.printStackTrace(); } } System.out.println("线程:" + threadName + ",占锁胜利!★★★"); } // 是否是以后线程占有锁,同时将超时工夫从新设置,这个很重要,同样也是原子操作 private boolean isHeldByCurrentThread() { List<byte[]> keys = Arrays.asList(encode(lock_key)); List<byte[]> args = Arrays.asList(encode(APP_ID + String.valueOf(threadId)), encode(String.valueOf(1))); long eval = (Long) jedis.eval(encode(addLockLife_lua), keys, args); return eval == 1; } private boolean tryLock(String uuid) { SetParams setParams = new SetParams(); setParams.ex(1); // 超时工夫1s setParams.nx(); // nx String response = jedis.set(lock_key, APP_ID + String.valueOf(threadId), setParams); // 转换为redis命令就是:set zjt-key xxx ex 1 nx return "OK".equals(response); } public void unlock(String uuid) { List<byte[]> keys = Arrays.asList(encode(lock_key)); List<byte[]> args = Arrays.asList(encode(APP_ID + String.valueOf(threadId))); // 应用lua脚本,保障原子性 long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args); if (eval == 1) { System.out.println("线程:" + threadName + " 开释锁胜利!☆☆☆"); } else { System.out.println("线程:" + threadName + " 开释锁失败!该线程未持有锁!!!"); } } private byte[] encode(String param) { return param.getBytes(); }
5-3 APP_ID(实例惟一标识) + ThreadId 还是 UUID 好呢?
持续听我说,如果 A线程执行逻辑两头开启了一个子线程执行工作,这个子线程工作中也须要重入这把锁,因为子线程获取到的线程 id不一样,导致重入失败。那意味着须要将这个秘钥持续传递给子线程,JUC中 InheritableThreadLocal 派上用场,然而感觉怪怪的,因为线程间传递的是父线程的 id。
微服务中多服务间调用的话能够借用零碎本身有的 traceId作为秘钥即可。比方sgm中的traceId 或者 利用RPC框架的隐式传参
「至于抉择哪种 value的形式,依据理论的零碎设计 + 业务场景,抉择最合适的即可,没有最好,只有最合适。」
5-4、锁重入的超时工夫怎么设置?
留神,咱们上边的次要注意力在怎么重入进去,而咱们这是分布式锁,要思考的事件还有很多,重入进去后,超时工夫轻易设吗?
比方说 A线程在锁办法中调用了 x()办法,而 x()办法中也有获取锁的逻辑,如果 A线程获取锁后,执行过程中,到 x()办法时,这把锁是要重入进去的,然而请留神,这把锁的超时工夫如果小于第一次上锁的工夫,比方说 A线程设置的超时工夫是 1s,在 100ms的时候执行到 x()办法中,而 x()办法中设置的超时工夫是 100ms,那么意味着 100ms之后锁就开释了,而这个时候我的 A线程的主办法还没有执行完呢!却被重入锁设置的工夫搞坏了!这个怎么搞?
如果说我在内存中设置一个这把锁设置过的最大的超时工夫,重入的时候判断下传进来的工夫,我重入时 expire的时候始终设置成最大的工夫,而不是由重入锁随便升高锁工夫导致上一步的主锁呈现问题
放在内存中行吗?咱们上边举例中,调用的 x()办法是在一个 JVM中,如果是调用近程的一个 RPC服务呢(像这种调用的话就须要将秘钥value通过 RpcContext传递过来了)到另一个节点的服务中进行锁重入,这个工夫仍然是要用以后设置过锁的最大工夫的,所以这个最大的工夫要存在 redis中而非 JVM内存中
通过这一步的剖析,咱们的重入 lua脚本就批改为这样了:
ADD_LOCK_LIFE("if redis.call(\"get\", KEYS[1]) == ARGV[1]\n" + // 判断是否是锁持有者 "then\n" + " local thisLockMaxTimeKeepKey=KEYS[1] .. \":maxTime\"\n" + // 记录锁最大工夫的key是:锁名字:maxTime " local nowTime=tonumber(ARGV[2])\n" + // 以后传参进来的time " local maxTime=redis.call(\"incr\", thisLockMaxTimeKeepKey)\n" + // 取出以后锁设置的最大的超时工夫,如果这个放弃工夫的key不存在返回的是字符串nil,这里为了lua脚本的易读性,用incr操作,这样读出来的都是number类型的操作 " local bigerTime=maxTime\n" + // 长期变量bigerTime=maxTime " if nowTime>maxTime-1\n" + // 如果传参进来的工夫>记录的最大工夫 " then\n" + " bigerTime=nowTime\n" + // 则更新bigerTime " redis.call(\"set\", thisLockMaxTimeKeepKey, tostring(bigerTime))\n" + // 设置超时工夫为最大的time,是最平安的 " else \n" + " redis.call(\"decr\", thisLockMaxTimeKeepKey)\n" + // 以后传参time<maxTime,将方才那次incr减回来 " end\n" + " return redis.call(\"expire\", KEYS[1], tostring(bigerTime))\n" + // 从新设置超时工夫为以后锁过的最大的time "else\n" + " return 0\n" + "end"),
其实,还有另外一种计划比较简单,就是锁的超时工夫=第一次上锁的工夫+前面所有重入锁的工夫。也就是(expire = 主ttl + 重入exipre),这种计划是放大的思维,一放大就又有上边提到过的一个问题:expire太大怎么办,参考上边。
5-5、重入锁的办法中间接执行 unlock?思考重入次数
A线程执行一共须要500ms,执行中须要调用 x()办法,x()办法中有一个重入锁,执行用了 50ms,而后执行完后,x()办法的 finally{} 块中将锁进行开释。
为啥能开释掉?因为秘钥我有,匹配胜利了我就间接开释了。
这当然是有问题的,所以咱们要通过锁重入次数来进行开释锁时候的判断,也就是说上锁的时候须要多保护一个 key来保留以后锁的重入次数,如果执行开释锁时,先进行重入次数 -1,-1后如果是0,能够间接 del,如果>0,阐明还有重入的锁在,不能间接 del。
5-6 思考如何存储锁的属性(锁的key 重入次数key 最大超时工夫key)?
目前为止,算上上一步中设置最大超时工夫的key,加上这一步重入次数的key,加上锁自身的key,曾经有3个key,须要留神的事件是,这三个key的超时工夫是都要设置的!为什么?如果说重入次数的 key没有设置超时工夫,服务A节点中在一个JVM中重入了5次后,调用一次 RPC服务,RPC服务中同样重入锁,此时,锁重入次数是 6,这个时候A服务宕机,就意味着无论怎样,这把锁不可能开释了,这个分布式锁提供的残缺能力,全线不可用了!
所以,这几个 key是要设置超时工夫的!怎么设置?我上一个锁要保护这么多 key的超时工夫?太简单了吧,多则乱,则容易出问题。怎么办?咱们想一下,是不是最大超时工夫的 key和重入次数的 key,都从属于锁,它们都是锁的属性,如果锁不在了,谈它们就毫无意义,这个时候用什么存储呢?redis的 hash数据结构,就能够做,key是锁,里边的 hashKey别离是锁的属性, hashValue是属性值,超时工夫只设置锁自身 key就能够了。这个时候,咱们的锁的数据结构就要扭转一下了。
6、如何解决过期工夫确定和业务执行时长不确定性的问题:看门狗机制
3-2中设置超时工夫那里,咱们预估锁办法执行工夫是 200ms,咱们放大 5倍后,设置超时工夫是 1s(过期工夫确定)。假想一下,如果生产环境中,锁办法中的 IO操作,极其状况下超时重大,比方说 IO就耗费了 2s(业务执行时长不确定),那就意味着,在这次 IO还没有完结的时候,我这把锁曾经到期开释掉了,就意味着别的线程趁虚而入,分布式锁解体!
咱们要做的是一把分布式锁,想要的目标是同一时刻只有一个线程持有锁,作为服务而言,这个锁当初不论是被哪个线程上锁胜利了,我服务应该保障这个线程执行的安全性,怎么办?锁续命(看门狗机制)。什么意思,一旦这把锁呈现了上锁操作,就意味着这把锁开始投入使用,这时我的服务中须要有一个 daemon线程定时去守护我的锁的安全性,怎么守护?比如说锁超时工夫设置的是 1s,那么我这个定时工作是每隔 300ms去 redis服务端做一次查看,如果我还持有,你就给我续命,就像 session会话的沉闷机制一样。看个例子,我上锁时候超时工夫设置的是 1s,理论办法执行工夫是 3s,这两头我的定时线程每隔 300ms就会去把这把锁的超时工夫从新设置为 1s,每隔 300ms一次,胜利将锁续命胜利。
public class RedisLockIdleThreadPool { private String threadAddLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + "then\n" + " return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + "else\n" + " return 0\n" + "end"; private volatile ScheduledExecutorService scheduledThreadPool; public RedisLockIdleThreadPool() { if (scheduledThreadPool == null) { synchronized (this) { if (scheduledThreadPool == null) { scheduledThreadPool = Executors.newSingleThreadScheduledExecutor(); // 我这样创立线程池是为了代码的易读性,大家务必应用ThreadPoolExecutor去创立 scheduledThreadPool.scheduleAtFixedRate(() -> { addLife(); }, 0, 300, TimeUnit.MILLISECONDS); } } } } private void addLife() { // ... 省略jedis的初始化过程 List<byte[]> keys = Arrays.asList(RedisLock.lock_key.getBytes()); List<byte[]> args = Arrays.asList(String.valueOf(1).getBytes()); jedis.eval(threadAddLife_lua.getBytes(), keys, args); } }
这就行吗?还不行!
为啥?想一下,如果每个服务中都像这样去续命锁,如果说A服务还在执行过程中的时候,还没有执行完,就是说还没有手动开释锁的时候,宕机,此时 redis中锁还在有效期。服务B 也始终在续命这把锁,此时这把锁始终在续命,然而 B的这个续命始终续的是 A过后设的锁,这不是扯吗?我本人在一直续命,导致我的服务上始终获取不到锁,实际上 A曾经宕机了呀!该开释了,不应该去续命了,这不是我服务 B该干的活!
续命的前提是,得判断是不是以后过程持有的锁,也就是咱们的 APP_ID,如果不是就不进行续命。
续命锁的 lua脚本产生扭转,如下:
THREAD_ADD_LIFE("local v=redis.call(\"get\", KEYS[1]) \n" + // get key "if v==false \n" + // 如果不存在key,读出后果v是false "then \n" + // 不存在不解决 "else \n" + " local match = string.find(v, ARGV[1]) \n" + // 存在,判断是否能和APP_ID匹配,匹配不上时match是nil " if match==\"nil\" \n" + " then \n" + " else \n" + " return redis.call(\"expire\", KEYS[1], ARGV[2]) \n" + // 匹配上了返回的是索引地位,如果匹配上了意味着就是以后过程占有的锁,就延长时间 " end \n" + "end")
6-1 锁在我手里,我挂了,这... 没救。只能期待锁超时开释
即使设置了一个很正当的 expire,比方 10s,然而线上如果真呈现了A节点刚拿到锁就宕机了,那其余节点也只能干等10s,之后能力拿到锁。次要还是业务能不能承受。而如果是 To C的业务中,大部分场景无奈承受的,因为可能会导致用户散失。所以咱们须要另外一个监控服务,定时去监控 redis中锁的获得者的衰弱状态,如果获取者超过n次无奈通信,由监控服务负责将锁摘除掉,让别的线程持续去获取到锁去干活。