如何優雅地用Redis實現分布式鎖
https://mp.weixin.qq.com/s?__biz=MzAxNjM2MTk0Ng==&mid=2247484976&idx=2&sn=a0b6771f0b4e471c710f8cd51c243971&chksm=9bf4b685ac833f936f3722a795ae202a3be37a3fb57332393e2eec3bbf8b34c4705d5b14a964&mpshare=1&scene=1&srcid=0919f7t2duWiufIOPFVMCzTH#rd
什麽是分布式鎖
在學習Java多線程編程的時候，鎖是一個很重要也很基礎的概念，鎖可以看做是多線程情況下訪問共享資源的一種線程同步機制。這是對於單進程應用而言的，即所有線程都在同一個JVM進程裏的時候，使用Java語言提供的鎖機制可以起到對共享資源進行同步的作用。如果分布式環境下多個不同線程需要對共享資源進行同步，那麽用Java的鎖機制就無法實現了，這個時候就必須借助分布式鎖來解決分布式環境下共享資源的同步問題。分布式鎖有很多種解決方案，今天我們要講的是怎麽使用緩存數據庫Redis來實現分布式鎖。
Redis分布式鎖方案一
使用Redis實現分布式鎖最簡單的方案是在獲取鎖之前先查詢一下以該鎖為key對應的value存不存在，如果存在，則說明該鎖被其他客戶端獲取了，否則的話就嘗試獲取鎖，獲取鎖的方法很簡單，只要以該鎖為key，設置一個隨機的值就行了。比如，我們有一批任務需要由多個分布式線程處理，每個任務都有一個taskId，為了保證每個任務只被執行一次，在工作線程執行任務之前，先獲取該任務的鎖，鎖的key可以為taskId。因此，獲取鎖的過程可以用如下偽代碼實現：
function boolean getLock(taskId){ if(existsKey(taskId)){ return false; }else{
setKey(taskId);
return true;
}
}
上述就是最簡單的獲取鎖的方案了，但是大家可以想想這個方案有什麽問題呢？有沒有什麽潛在的坑？在分析這種方案的優缺點之前，先說一下獲取鎖後我們一般是怎麽使用鎖，並且又是如何釋放鎖的，以Java語言為例，我們一般獲取鎖後會將釋放鎖的代碼放在finally塊中，這樣做的好處是即使在使用鎖的過程中出現異常，也能順利將鎖釋放掉。用偽代碼描述如下：
boolean lock=false;
try{
lcok=getLock(taskId); //獲取鎖
if(lock){
doSomething(); //業務邏輯
}
}finally{
if(lock){
releaseLock(taskId); //釋放鎖
}
}
其中，getLock方法的偽代碼上文已經給出，releaseLock方法是釋放鎖的方法，在該方案中，只是簡單地刪除掉key，就不給出偽代碼了。
上述使用鎖的代碼咋一看是沒有什麽問題的，學過Java的人都知道，在try...finally...代碼塊中，即使try代碼塊中拋出異常，最終也會執行finally代碼塊，然而這樣就能保證鎖一定會被釋放嗎？考慮這樣一種情況：代碼執行到doSomething()方法的時候，服務器宕機了，這個時候finally代碼塊就沒法被執行了，因此在這種情況下，該鎖不會被正常釋放，在上述案例中，可能會導致任務漏算。因此，這種方案的第一個問題是會出現鎖無法正常釋放的風險，解決這個問題的方法也很簡單，Redis設置key的時候可以指定一個過期時間，只要獲取鎖的時候設置一個合理的過期時間，那麽即使服務器宕機了，也能保證鎖被正確釋放。
該方案的另外一個問題是，獲取到的鎖不一定是排他鎖，也就是說同一把鎖同一時間可能被不同客戶端獲取到。仔細分析一下getLock方法，該方法並不是原子性的，當一個客戶端檢查到某個鎖不存在，並在執行setKey方法之前，別的客戶端可能也會檢查到該鎖不存在，並也會執行setKey方法，這樣一來，同一把鎖就有可能被不同的客戶端獲取到了。
既然這種方案有以上缺點，那麽該如何改進呢？且聽我慢慢道來。
Redis分布式鎖方案二
上一小節的方案有2個缺點，一個是獲取的鎖可能無法釋放，另一個是同一把鎖在同一時間可能被不同線程獲取到。通過查看Redis文檔，可以找到Redis提供了一個只有在某個key不存在的情況下才會設置key的值的原子命令，該命令也能設置key值過期時間，因此使用該命令，不存在上述方案出現的問題，該命令為：
SET my_key my_value NX PX milliseconds
其中，NX表示只有當鍵key不存在的時候才會設置key的值，PX表示設置鍵key的過期時間，單位是毫秒。
如此一來，獲取鎖的過程可以用如下偽代碼描述：
function boolean getLock(taskId,timeout){ return setKeyOnlyIfNotExists(taskId,timeout); }
其中，setKeyOnlyIfNotExists方法表示的是原子命令SET my_key my_value NX PX milliseconds。
如此一來，獲取鎖的代碼應該就沒什麽問題了，但是這種方案還是會有其他問題。大家再仔細研究下釋放鎖的代碼。因為現在我們設置key的時候也設置了過期時間，所以原來的釋放鎖的代碼現在看來就有問題了。考慮這樣一種情況：客戶端A獲取鎖的時候設置了key的過期時間為2秒，然後客戶端A在獲取到鎖之後，業務邏輯方法doSomething執行了3秒（大於2秒），當執行完業務邏輯方法的時候，客戶端A獲取的鎖已經被Redis過期機制自動釋放了，因此客戶端A在獲取鎖經過2秒之後，該鎖可能已經被其他客戶端獲取到了。當客戶端A執行完doSomething方法之後接下來就是執行releaseLock方法釋放鎖了，由於前面說了，該鎖可能已經被其他客戶端獲取到了，因此這個時候釋放鎖就有可能釋放的是其他客戶端獲取到的鎖。
Redis分布式鎖方案三
既然方案二可能會出現釋放了別的客戶端申請的鎖的問題，那麽該如何進行改進呢？有一個很簡單的方法是，我們設置key的時候，將value設置為一個隨機值r，當釋放鎖，也就是刪除key的時候，不是直接刪除，而是先判斷該key對應的value是否等於先前設置的隨機值，只有當兩者相等的時候才刪除該key，由於每個客戶端產生的隨機值是不一樣的，這樣一來就不會誤釋放別的客戶端申請的鎖了。新的釋放鎖的方案用偽代碼描述如下：
function void releaseLock(taskId,random_value){ if(getKey(taskId)==random_value){ deleteKey(taskId); } }
其中，getKey方法就是Redis的查詢key值的方法，deleteKey就是Redis的刪除key值的方法，在此不給出偽代碼了。
那麽這種方案就沒有問題了嗎？很遺憾地說，這種方案也是有問題的。原因在於上述釋放鎖的操作不是原子性的，不是原子性操作意味著當一個客戶端執行完getKey方法並在執行deleteKey方法之前，也就是在這2個方法執行之間，其他客戶端是可以執行其他命令的。考慮這樣一種情況，在客戶端A執行完getKey方法，並且該key對應的值也等於先前的隨機值的時候，接下來客戶端A將會執行deleteKey方法。假設由於網絡或其他原因，客戶端A執行getKey方法之後過了1秒鐘才執行deleteKey方法，那麽在這1秒鐘裏，該key有可能也會因為過期而被Redis清除了，這樣一來另一個客戶端，姑且稱之為客戶端B，就有可能在這期間獲取到鎖，然後接下來客戶端A就執行到deleteKey方法了，如此一來就又出現誤釋放別的客戶端申請的鎖的問題了。
Redis分布式鎖方案四
既然方案三的問題是因為釋放鎖的方法不是原子操作導致的，那麽我們只要保證釋放鎖的代碼是原子性的就能解決該問題了。很遺憾的是，查閱Redis開發文檔，並沒有發現相關的原子操作。不過幸運的是，在Redis中執行原子操作不止有通過官方提供的命令的方式，還有另外一種方式，就是Lua腳本。因此，方案三中的釋放鎖的代碼可以用以下Lua腳本來實現：
if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end
其中ARGV[1]表示設置key時指定的隨機值。
由於Lua腳本的原子性，在Redis執行該腳本的過程中，其他客戶端的命令都需要等待該Lua腳本執行完才能執行，所以不會出現方案三所說的問題。至此，使用Redis實現分布式鎖的方案就相對完善了。
總結
上述分布式鎖的實現方案中，都是針對單節點Redis而言的，然而在生產環境中，我們使用的通常是Redis集群，並且每個主節點還會有從節點。由於Redis的主從復制是異步的，因此上述方案在Redis集群的環境下也是有問題的。關於在Redis集群中如何優雅地實現分布式鎖，後續再寫文章詳述。
如何優雅地用Redis實現分布式鎖