從零到一帶你手寫基於Redis的分散式鎖框架
1.分散式鎖緣由
學習程式設計初期，我們做的諸如教務系統、成績管理系統大多是單機架構，單機架構在處理併發的問題上一般是依賴於JDK內建的併發程式設計類庫，如synchronize關鍵字、Lock類等。隨著業務以及需求的提高，單機架構不再滿足我們的要求，這個時候我們不免要進行業務上的分離，例如基於Maven進行多模組開發。業務與業務分離之後，遇到的首要問題就是業務之間如何進行通訊，相信會有不少讀者瞭解諸如Dubbo、SpringCloud之類的RPC框架，但這些RPC框架並沒有自帶處理分散式併發問題的功能，所以，分散式併發問題還需要我們自己去實現分散式鎖。
2.分散式鎖條件
為了確保分散式鎖可用，我們至少要確保鎖的實現同時滿足以下四個條件：
- 互斥性。在任意時刻，只有一個客戶端能持有鎖。
- 不會發生死鎖。即使有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其他客戶端能加鎖。
- 具有容錯性。只要大部分的Redis節點正常執行，客戶端就可以加鎖和解鎖。
- 解鈴還須繫鈴人。加鎖和解鎖必須是同一個客戶端，客戶端自己不能把別人加的鎖給解了。
3.分散式鎖方式
分散式鎖一般有三種實現方式：
- 資料庫樂觀鎖
- 基於Redis的分散式鎖
- 基於Zookeeper的分散式鎖
下面我按個提一下這三種方式的大致實現思路。
3.1 資料庫樂觀鎖
資料庫樂觀鎖的實現方式是先使用SELECT語句查詢某欄位的值（版本號），該欄位即理解為要獲取的分散式鎖。然後在使用UPDATE語句對正常業務資料進行更新，在UPDATE語句執行時一定要用WHERE條件對版本號進行判斷，若版本號在這段時間內並沒有發生變化則該語句預設執行成功，否則迴圈執行即可。
示例程式碼：
select (status,version) from goods where id=#{id}
update goods set status=2,version=version+1 where id=#{id} and version=#{version};3.2 基於Zookeeper的分散式鎖
基於Zookeeper實現分散式鎖的演算法思路大致如下假設鎖空間的根節點為/lock：
- 客戶端連線zookeeper，並在/lock下建立臨時的且有序的子節點，第一個客戶端對應的子節點為/lock/lock-0000000000，第二個為/lock/lock-0000000001，以此類推。
- 客戶端獲取/lock下的子節點列表，判斷自己建立的子節點是否為當前子節點列表中序號最小的子節點，如果是則認為獲得鎖，否則監聽/lock的子節點變更訊息，獲得子節點變更通知後重復此步驟直至獲得鎖。
- 執行業務程式碼。
- 完成業務流程後，刪除對應的子節點釋放鎖。
3.3 基於Redis的分散式鎖
基於Redis的分散式鎖實現是基於Redis自帶的 setnx 命令。該命令只有在要設定的欄位不存在的情況下才能設定成功，也就是獲得分散式鎖，否則失敗。為了防止客戶端異常導致的鎖未釋放問題，還需要對該欄位設定過期時間。
本文將基於Redis分散式鎖的實現思路設計一個spring-boot-starter-redis-lock框架。
核心程式碼如下：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
@Component
public class RedisLock {
@Autowired
private StringRedisTemplate template;
@Autowired
private DefaultRedisScript<Long> redisScript;
private static final Long RELEASE_SUCCESS = 1L;
private long timeout = 3000;
public boolean lock(String key, String value) {
//執行set命令
Boolean absent = template.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.MILLISECONDS);//1
//其實沒必要判NULL，這裡是為了程式的嚴謹而加的邏輯
if (absent == null) {
return false;
}
//是否成功獲取鎖
return true;
}
public boolean unlock(String key, String value) {
//使用Lua指令碼：先判斷是否是自己設定的鎖，再執行刪除
Long result = template.execute(redisScript, Arrays.asList(key,value));
//返回最終結果
return RELEASE_SUCCESS.equals(result);
}
public void setTimeout(long timeout) {
this.timeout = timeout;
}
@Bean
public DefaultRedisScript<Long> defaultRedisScript() {
DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>();
defaultRedisScript.setResultType(Long.class);
defaultRedisScript.setScriptText("if redis.call('get', KEYS[1]) == KEYS[2] then return redis.call('del', KEYS[1]) else return 0 end");
return defaultRedisScript;
}
}執行上面的setIfAbsent()方法就只會導致兩種結果：1. 當前沒有鎖（key不存在），那麼就進行加鎖操作，並對鎖設定個有效期，同時value表示加鎖的客戶端。2. 已有鎖存在，不做任何操作。
回顧上面提到的分散式鎖的四個條件，在任意時刻，該程式碼都能保證只有一個客戶端能持有鎖，並且每一個分散式鎖都加了過期時間，保證不會出現死鎖，容錯性暫時不考慮的話，加鎖和解鎖通過key保證了對多個客戶端而言都是同一把鎖，value的作用則是保證對同一把鎖的加鎖和解鎖操作都是同一個客戶端。
4.為什麼上述方案不夠好
為了理解我們想要提高的到底是什麼，我們先看下當前大多數基於Redis的分散式鎖三方庫的現狀。 用Redis來實現分散式鎖最簡單的方式就是在例項裡建立一個鍵值，創建出來的鍵值一般都是有一個超時時間的（這個是Redis自帶的超時特性），所以每個鎖最終都會釋放（參見前文屬性2）。而當一個客戶端想要釋放鎖時，它只需要刪除這個鍵值即可。 表面來看，這個方法似乎很管用，但是這裡存在一個問題：在我們的系統架構裡存在一個單點故障，如果Redis的master節點宕機了怎麼辦呢？有人可能會說：加一個slave節點！在master宕機時用slave就行了！但是其實這個方案明顯是不可行的，因為這種方案無法保證第1個安全互斥屬性，因為Redis的複製是非同步的。 總的來說，這個方案裡有一個明顯的競爭條件（race condition），舉例來說：
- 客戶端A在master節點拿到了鎖。
- master節點在把A建立的key寫入slave之前宕機了。
- slave變成了master節點
- B也得到了和A還持有的相同的鎖（因為原來的slave裡還沒有A持有鎖的資訊）
當然，在某些特殊場景下，前面提到的這個方案則完全沒有問題，比如在宕機期間，多個客戶端允許同時都持有鎖，如果你可以容忍這個問題的話，那用這個基於複製的方案就完全沒有問題，否則的話我還是建議你對上述方案進行改進。比如，考慮使用Redlock演算法。
5.Redlock演算法
在分散式版本的演算法裡我們假設我們有N個Redis master節點，這些節點都是完全獨立的，我們不用任何複製或者其他隱含的分散式協調演算法。我們已經描述瞭如何在單節點環境下安全地獲取和釋放鎖。因此我們理所當然地應當用這個方法在每個單節點裡來獲取和釋放鎖。在我們的例子裡面我們把N設成5，這個數字是一個相對比較合理的數值，因此我們需要在不同的計算機或者虛擬機器上執行5個master節點來保證他們大多數情況下都不會同時宕機。一個客戶端需要做如下操作來獲取鎖：
- 獲取當前時間（單位是毫秒）。
- 輪流用相同的key和隨機值在N個節點上請求鎖，在這一步裡，客戶端在每個master上請求鎖時，會有一個和總的鎖釋放時間相比小的多的超時時間。比如如果鎖自動釋放時間是10秒鐘，那每個節點鎖請求的超時時間可能是5-50毫秒的範圍，這個可以防止一個客戶端在某個宕掉的master節點上阻塞過長時間，如果一個master節點不可用了，我們應該儘快嘗試下一個master節點。
- 客戶端計算第二步中獲取鎖所花的時間，只有當客戶端在大多數master節點上成功獲取了鎖（在這裡是3個），而且總共消耗的時間不超過鎖釋放時間，這個鎖就認為是獲取成功了。
- 如果鎖獲取成功了，那現在鎖自動釋放時間就是最初的鎖釋放時間減去之前獲取鎖所消耗的時間。
- 如果鎖獲取失敗了，不管是因為獲取成功的鎖不超過一半（N/2+1)還是因為總消耗時間超過了鎖釋放時間，客戶端都會到每個master節點上釋放鎖，即便是那些他認為沒有獲取成功的鎖。
本文程式碼倉庫：https://github.com/yueshutong/spring-boot-starter-redis-lock
參考文章：https://www.cnblogs.com/ironPhoenix/p/6048467.html