共计 12929 个字符，预计需要花费 33 分钟才能阅读完成。
作者：京东保险 张江涛
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.html
SET key value [EX seconds] [PX milliseconds] [NX|XX]不要在代码里边分两次调用：
set k v
exipre k time3-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 次无奈通信，由监控服务负责将锁摘除掉，让别的线程持续去获取到锁去干活。
