Redis避坑指南：為什麼要有分散式鎖？
作者：京東保險 張江濤
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 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次無法通訊，由監控服務負責將鎖摘除掉，讓別的執行緒繼續去獲取到鎖去幹活。