Redis避坑指南：為什麼要有分散式鎖？

来源:https://www.cnblogs.com/Jcloud/archive/2023/02/14/17118976.html
-Advertisement-
Play Games

JUC提供的鎖機制，可以保證在同一個JVM進程中同一時刻只有一個線程執行操作邏輯； 多服務多節點的情況下，就意味著有多個JVM進程，要做到這樣，就需要有一個中間人； 分散式鎖就是用來保證在同一時刻，僅有一個JVM進程中的一個線程在執行操作邏輯； 換句話說，JUC的鎖和分散式鎖都是一種保護系統資... ...


作者：京東保險 張江濤

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次無法通信，由監控服務負責將鎖摘除掉，讓別的線程繼續去獲取到鎖去幹活。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 修改、獲取Linux內核版本信息的方法
    自己編譯的內核進行修改後為後續方便查詢是那個版本的系統。 所以每次更改內核後都需要修改一下版本信息， 又因為內核一般是不變的為了區分所以增加到擴展版本上。 操作環境： 硬體是全志 V3S Linux內核是3.4 修改的方法： 方法一： 一個在menuconfig中進行增加 打開menuconfig ...
  • nRF5340開發——晶振篇
    Vim 簡介{#vim-簡介} Vim 是 Linux 系統上的最著名的文本/ 代碼編輯器,也是早年的 Vi編輯器的加強版，而 gVim 則是其 Windows 版。它的最大特色是完全使用鍵盤命令進行編輯，脫離了滑鼠操作雖然使得入門變得困難,但上手之後鍵盤流的各種巧妙組合操作卻能帶來極為大幅的效率提 ...
  • redis georadius源碼分析與性能優化
    原文地址: https://blog.fanscore.cn/a/51/ 背景 最近接到一個需求，開發中使用了redis georadius命令取附近給定距離內的點。完工後對服務進行壓測後發現georadius的性能比預期要差，因此我分析了georadius的源碼，並對原始的實現方案進行了優化，總結 ...
  • sql語氣查詢去重的兩種方法、sql的模糊查詢、sql中的日期函數、mysql字元串截取之substring_index、sql中字元串的切割、截取、刪除、替換、計算字元串長度函數、返回某個日期之間的解法
    sql語氣查詢去重的兩種方法 以下兩種都可以進行去重查詢，區別是： 用distinct去重，只能查詢到去重的屬性那一列，無法查詢其他欄位 用group by分組查詢，可以根據需求查詢對應的其他欄位，推薦用group by 第一種：使用distinct(關鍵詞distinct用於返回唯一不同的值) s ...
  • MRS+LakeFormation：打造一站式湖倉，釋放數據價值
    摘要：華為LakeFormation是企業級的一站式湖倉構建服務。 本文分享自華為雲社區《華為雲MRS支持LakeFormation能力,打造一站式湖倉，釋放數據價值】》，作者：breakDawn 。 1 背景 1.1 數倉和數據湖的概念 數據分析技術在2010~2019年間，以湖倉兩層架構技術作為 ...
  • 新增或修改mysql表欄位--可重覆執行腳本
    想必大家在版本更新迭代中，不免需要對資料庫表的欄位進行修改欄位類型或者長度的情況， 如果是修改存儲過程或者自定義函數, 那還是很好改的, 不用擔心表功能收到影響. 如果是改欄位呢? 首先要判斷欄位是不是已經在系統裡面, 沒有的話, 需要新增上去, 有的話, 需不需要修改? 其次, 腳本如果報錯, 是 ...
  • 倒帶ChunJun，同心前行｜2022年度回顧&2023年共建規劃
    ChunJun是一個開始於2018年的批流一體數據集成框架項目，原名FlinkX。2022年2月22日，在FlinkX進行初版開源的整整四年後，技術團隊決定對FlinkX進行整體升級，並更名為ChunJun，希望為大家真正提供一個穩定、高效、易用的批流一體的數據集成框架。 2022年的ChunJun ...
  • GaussDB(DWS)性能調優：indexscan導致的性能問題識別與優化
    摘要：通常跑批加工場景下，都是大數量做關聯操作，通常不建議使用索引。有些時候因為計劃誤判導致使用索引的可能會導致嚴重的性能問題。本文從一個典型的索引導致性能的場景重發，剖析此類問題的特征，定位方法和解決方法 本文分享自華為雲社區《GaussDB(DWS)性能調優：indexscan導致的性能問題識別 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類