從零到一手寫基於Redis的分佈式鎖框架

1.分佈式鎖原因

學習編程初期，咱們作的諸如教務系統、成績管理系統大可能是單機架構，單機架構在處理併發的問題上通常是依賴於JDK內置的併發編程類庫，如synchronize關鍵字、Lock類等。隨着業務以及需求的提升，單機架構再也不知足咱們的要求，這個時候咱們難免要進行業務上的分離，例如基於Maven進行多模塊開發。業務與業務分離以後，遇到的首要問題就是業務之間如何進行通訊，相信會有很多讀者瞭解諸如Dubbo、SpringCloud之類的RPC框架，但這些RPC框架並無自帶處理分佈式併發問題的功能，因此，分佈式併發問題還須要咱們本身去實現分佈式鎖。

2.分佈式鎖條件

爲了確保分佈式鎖可用，咱們至少要確保鎖的實現同時知足如下四個條件：

1. 互斥性。在任意時刻，只有一個客戶端能持有鎖。
2. 不會發生死鎖。即便有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其餘客戶端能加鎖。
3. 具備容錯性。只要大部分的Redis節點正常運行，客戶端就能夠加鎖和解鎖。
4. 解鈴還須繫鈴人。加鎖和解鎖必須是同一個客戶端，客戶端本身不能把別人加的鎖給解了。

3.分佈式鎖方式

分佈式鎖通常有三種實現方式：

1. 數據庫樂觀鎖
2. 基於Redis的分佈式鎖
3. 基於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：

1. 客戶端鏈接zookeeper，並在/lock下建立臨時的且有序的子節點，第一個客戶端對應的子節點爲/lock/lock-0000000000，第二個爲/lock/lock-0000000001，以此類推。
2. 客戶端獲取/lock下的子節點列表，判斷本身建立的子節點是否爲當前子節點列表中序號最小的子節點，若是是則認爲得到鎖，不然監聽/lock的子節點變動消息，得到子節點變動通知後重復此步驟直至得到鎖。
3. 執行業務代碼。
4. 完成業務流程後，刪除對應的子節點釋放鎖。

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），舉例來講：

1. 客戶端A在master節點拿到了鎖。
2. master節點在把A建立的key寫入slave以前宕機了。
3. slave變成了master節點
4. B也獲得了和A還持有的相同的鎖（由於原來的slave裏尚未A持有鎖的信息）

固然，在某些特殊場景下，前面提到的這個方案則徹底沒有問題，好比在宕機期間，多個客戶端容許同時都持有鎖，若是你能夠容忍這個問題的話，那用這個基於複製的方案就徹底沒有問題，不然的話我仍是建議你對上述方案進行改進。好比，考慮使用Redlock算法。

5.Redlock算法

在分佈式版本的算法裏咱們假設咱們有N個Redis master節點，這些節點都是徹底獨立的，咱們不用任何複製或者其餘隱含的分佈式協調算法。咱們已經描述瞭如何在單節點環境下安全地獲取和釋放鎖。所以咱們理所固然地應當用這個方法在每一個單節點裏來獲取和釋放鎖。在咱們的例子裏面咱們把N設成5，這個數字是一個相對比較合理的數值，所以咱們須要在不一樣的計算機或者虛擬機上運行5個master節點來保證他們大多數狀況下都不會同時宕機。一個客戶端須要作以下操做來獲取鎖：

1. 獲取當前時間（單位是毫秒）。
2. 輪流用相同的key和隨機值在N個節點上請求鎖，在這一步裏，客戶端在每一個master上請求鎖時，會有一個和總的鎖釋放時間相比小的多的超時時間。好比若是鎖自動釋放時間是10秒鐘，那每一個節點鎖請求的超時時間多是5-50毫秒的範圍，這個能夠防止一個客戶端在某個宕掉的master節點上阻塞過長時間，若是一個master節點不可用了，咱們應該儘快嘗試下一個master節點。
3. 客戶端計算第二步中獲取鎖所花的時間，只有當客戶端在大多數master節點上成功獲取了鎖（在這裏是3個），並且總共消耗的時間不超過鎖釋放時間，這個鎖就認爲是獲取成功了。
4. 若是鎖獲取成功了，那如今鎖自動釋放時間就是最初的鎖釋放時間減去以前獲取鎖所消耗的時間。
5. 若是鎖獲取失敗了，不論是由於獲取成功的鎖不超過一半（N/2+1)仍是由於總消耗時間超過了鎖釋放時間，客戶端都會到每一個master節點上釋放鎖，即使是那些他認爲沒有獲取成功的鎖。

本文代碼倉庫：https://github.com/yueshutong/spring-boot-starter-redis-lock

參考文章：https://www.cnblogs.com/ironPhoenix/p/6048467.html