道阻且長，行則將至，請相信我，你一定會更優秀！

「 Redis 做分布式鎖，沒那么簡單，調整好心態，保證你有識訓」

本文我們主要聊 redis實作分布式鎖，別的不聊，先來三個問題熱熱身：

一個 setnx 就行了？還有人認為 incr 也可以？
再加個超時時間就行了？
你寫的分布式鎖，你確認你敢在生產環境用嗎？

熱身完畢，下邊我和你一起學習下 Redis分布式鎖到底該怎么玩？

1、為什么要有分布式鎖？

JUC提供的鎖機制，可以保證在同一個JVM行程中同一時刻只有一個執行緒執行操作邏輯；
多服務多節點的情況下，就意味著有多個JVM行程，要做到這樣，就需要有一個中間人；
分布式鎖就是用來保證在同一時刻，僅有一個JVM行程中的一個執行緒在執行操作邏輯；
換句話說，JUC的鎖和分布式鎖都是一種保護系統資源的措施，盡可能將并發帶來的不確定性轉換為同步的確定性；

2、先捋脈絡，再想風險，最后再寫代碼

當我們設計一個東西的時候，很多同學腦子里想到的第一件事就是代碼，代碼，聽我說，你一定要先思考，要做一根能思想的葦草，代碼是死的，三思而后行，

所以，一定要先在腦子里想，這把鎖，我要用它干什么，它要保證什么，有沒有什么意外情況，會存在什么風險，先全域看一下，別一下子鉆到里邊，想完了之后，然后一定要落地，絕對不可以紙上談兵，自己一定要把代碼寫出來，自己去測驗，去解決問題，看到底行不行，只有寫出來，你才能驗證你的想法，“實踐是檢驗真理的唯一標準”，

為了保證文章的易讀性，接下來，我將采用理論 + 代碼的形式，從整體到部分，從宏觀到微觀，帶你全面看透 Redis分布式鎖，

3、一步一步，看透 Redis 分布式鎖中的門道

我們一起捋一下，很多執行緒去上鎖，誰鎖成功誰就有權利執行操作邏輯，其他執行緒要么直接走搶鎖失敗的邏輯，要么自旋嘗試搶鎖；

比方說 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{} 中釋放鎖
    }
}

3-1、放在 finally{} 塊中就行了嗎？

如果在執行 try{} 中邏輯的時候，程式出現了 System.exit(0); 或者 finally{} 中執行例外，比方說連接不上 redis-server了；或者還未執行到 finally{}的時候，JVM行程掛掉了，服務宕機；這些情況都會導致沒有成功釋放鎖，別的執行緒一直拿不到鎖，怎么辦？如果我的系統因為一個節點影響，別的節點也都無法正常提供服務了，那我的系統也太弱了，所以我們想到必須要將風險降低，可以給鎖設定一個超時時間，比方說 1秒，即便發生了上邊的情況，那我的鎖也會在 1秒之后自動釋放，其他執行緒就可以獲取到鎖，接班干活了；
```
	 public static final String lock_key = "haolin-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 haolin-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、1min不是更安全嗎？請注意，不要鉆到這一個點里邊，你要顧全大局，多大算大？越大越好？無窮大？那不等于不設定超時時間嗎？同時，這個時間，你要想清楚，如果你設定 10s，果真發生了宕機，那意味著這 10s中間，你的這個分布式鎖的服務全部節點都是不可用的，這個和你的業務以及系統的可用性有掛鉤，你要去衡量，要慎重，

3-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 haolin-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();
	}

3-4、這個密語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 haolin-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已經變了，也就是說持有鎖的執行緒已經不是該執行緒了，此時再執行第二步的洗掉鎖操作，肯定是錯誤的了，

3-5、繼續，現在把思維先跳出來，想想？可重入怎么搞？

作為一把鎖，我們在使用 synchronized、ReentrantLock的時候是不是有可重入性？
那咱們這把分布式鎖該如何實作可重入呢？如果 A執行緒的鎖方法邏輯中呼叫了 x()方法，x()方法中也需要獲取這把鎖，按照這個邏輯，x()方法中的鎖應該重入進去即可，那是不是需要將剛才生成的這個 UUID秘鑰傳遞給 x()方法？怎么傳遞？引數？這就侵入業務代碼了，

3-6、能不侵入業務系統嗎？

我們主要是想給上鎖的 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 haolin-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();
	}

3-7、Thread-Id 真能行嗎？

不行，
想想，我們說一個 Thread的id是唯一的，是在同一個 JVM行程中，是在一個作業系統中，也就是在一個機器中，而現實是，我們的部署是集群部署，多個實體節點，那意味著會存在這樣一種情況，S1機器上的執行緒上鎖成功，此時鎖中秘鑰 value是執行緒id=1，如果說同一時間 S2機器中，正好執行緒id=1的執行緒嘗試獲得這把鎖，比對秘鑰發現成功，結果也重入了這把鎖，也開始執行邏輯，此時，我們的分布式鎖崩潰！怎么解決？我們只需要在每個節點中維護不同的標識即可，怎么維護呢？應用啟動的時候，使用 UUID生成一個唯一標識 APP_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 haolin-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();
	}

3-8、APP_ID + ThreadId 能解決下邊這個問題嗎？

是不是覺得有點意思了？
繼續聽我說，如果 A執行緒執行邏輯中間開啟了一個子執行緒執行任務，這個子執行緒任務中也需要重入這把鎖，因為子執行緒獲取到的執行緒 id不一樣，導致重入失敗，那意味著需要將這個秘鑰繼續傳遞給子執行緒，JUC中 InheritableThreadLocal 派上用場，但是感覺怪怪的，因為執行緒間傳遞的是父執行緒的 id，

「至于選擇哪種 value的方式，根據實際的系統設計 + 業務場景，選擇最合適的即可，沒有最好，只有最合適，」

3-9、豁然開朗？再來一招！

再一次把思路時鐘往回撥，回撥到設定超時時間那里，我們預估鎖方法執行時間是 200ms，我們放大 5倍后，設定超時時間是 1s，假想一下，如果生產環境中，鎖方法中的 IO操作，極端情況下超時嚴重，比方說 IO就消耗了 2s，那就意味著，在這次 IO還沒有結束的時候，我這把鎖已經到期釋放掉了，就意味著別的執行緒趁虛而入，分布式鎖崩潰！

3-10、搞了半天，鎖還是崩潰了？

跟著我的思路走，別放棄，
再一次把思維從現在的框框里跳出來，想一想，我們要做的是一把分布式鎖，想要的目的是同一時刻只有一個執行緒持有鎖，作為服務而言，這個鎖現在不管是被哪個執行緒上鎖成功了，我服務應該保證這個執行緒執行的安全性，怎么辦？鎖續命，什么意思，一旦這把鎖出現了上鎖操作，就意味著這把鎖開始投入使用，這時我的服務中需要有一個 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();
					
					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);
	}
	
}

4、風險！主從部署引來的問題 <master - slave>

哨兵主從部署的時候，會存在一個風險問題，因為 Redis默認的主從復制是異步的，那很自然可以想到一個問題，極端情況下，如果剛往 master節點寫入一個分布式鎖，而這個指令流還沒有來得及同步給任意一個 slave節點，此時，master節點宕機，其中一個 slave被哨兵選舉為 master，此時是沒有這個鎖的，別的執行緒再次來獲取鎖，又獲取鎖成功了，當然，這個概率極低，但是我們必須得承認這個風險的存在，本文對這塊不探究太細，后面我會和大家專門聊聊 Redis集群的那些事，

從 Redis官方檔案上摘抄如下（https://redis.io/topics/replication）：

Redis uses by default asynchronous replication, which being low latency and high performance, is the natural replication mode for the vast majority of Redis use cases.

譯文：Redis默認使用異步復制，低延遲和高性能，絕大多數的Redis服務使用自然復制模式，

完工，我建議你合上螢屏，自己在腦子里重新過一遍，每一步都在做什么，為什么要做，解決什么問題，想清楚之后，一定要，一定要自己親手來一遍代碼，

【下集預告】10分鐘拿下zookeeper 分布式鎖各種門道，會比 redis優秀嗎？下期見，

努力改變自己和身邊人的生活，

特別希望本文可以對您有所幫助，轉載請注明出處，感謝大家留言討論交流，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/238022.html

標籤：其他

上一篇：初識tomcat

下一篇：選擇排序和陣列

10分鐘精通 Redis分布式鎖中的各種門道