弄懂這些redis分布式鎖知識點，明天就去跟老板談漲薪！（建議收藏）

1 介紹

這篇博文講介紹如何一步步構建一個基于Redis的分布式鎖，會從最原始的版本開始，然后根據問題進行調整，最后完成一個較為合理的分布式鎖，

本篇文章會將分布式鎖的實作分為兩部分，一個是單機環境，另一個是集群環境下的Redis鎖實作，在介紹分布式鎖的實作之前，先來了解下分布式鎖的一些資訊，

開始之前，記得點贊收藏加關注哦 ，需要下載PDF版本和獲取更多知識點、面試題的朋友可以點一點下方鏈接免費領取

鏈接：點這里！！！ 799215493 暗號：CSDN

2 分布式鎖

2.1 什么是分布式鎖？

分布式鎖是控制分布式系統或不同系統之間共同訪問共享資源的一種鎖實作，如果不同的系統或同一個系統的不同主機之間共享了某個資源時，往往需要互斥來防止彼此干擾來保證一致性，

2.2 分布式鎖需要具備哪些條件

1. 互斥性：在任意一個時刻，只有一個客戶端持有鎖，
2. 無死鎖：即便持有鎖的客戶端崩潰或者其他意外事件，鎖仍然可以被獲取，
3. 容錯：只要大部分Redis節點都活著，客戶端就可以獲取和釋放鎖

2.3 分布式鎖的實作有哪些？

1. 資料庫
2. Memcached（add命令）
3. Redis（setnx命令）
4. Zookeeper（臨時節點）
5. 等等

3 單機Redis的分布式鎖

3.1 準備作業

3.1.1 定義常量類

public class LockConstants {
    public static final String OK = "OK";

    /** NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key if it already exist. **/
    public static final String NOT_EXIST = "NX";
    public static final String EXIST = "XX";

    /** expx EX|PX, expire time units: EX = seconds; PX = milliseconds **/
    public static final String SECONDS = "EX";
    public static final String MILLISECONDS = "PX";

    private LockConstants() {}
}

3.1.2 定義鎖的抽象類

抽象類RedisLock實作java.util.concurrent包下的Lock介面，然后對一些方法提供默認實作，子類只需實作lock方法和unlock方法即可，代碼如下

public abstract class RedisLock implements Lock {

    protected Jedis jedis;
    protected String lockKey;

    public RedisLock(Jedis jedis,String lockKey) {
        this(jedis, lockKey);
    }


    public void sleepBySencond(int sencond){
        try {
            Thread.sleep(sencond*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void lockInterruptibly(){}

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public boolean tryLock() {
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit){
        return false;
    }

}

3.2 最基礎的版本1

先來一個最基礎的版本，代碼如下

public class LockCase1 extends RedisLock {

    public LockCase1(Jedis jedis, String name) {
        super(jedis, name);
    }

    @Override
    public void lock() {
        while(true){
            String result = jedis.set(lockKey, "value", NOT_EXIST);
            if(OK.equals(result)){
                System.out.println(Thread.currentThread().getId()+"加鎖成功!");
                break;
            }
        }
    }

    @Override
    public void unlock() {
        jedis.del(lockKey);
    }
}

LockCase1類提供了lock和unlock方法，

其中lock方法也就是在reids客戶端執行如下命令

SET lockKey value NX

而unlock方法就是呼叫DEL命令將鍵洗掉，

好了，方法介紹完了，現在來想想這其中會有什么問題？

假設有兩個客戶端A和B，A獲取到分布式的鎖，A執行了一會，突然A所在的服務器斷電了（或者其他什么的），也就是客戶端A掛了，這時出現一個問題，這個鎖一直存在，且不會被釋放，其他客戶端永遠獲取不到鎖，如下示意圖

可以通過設定過期時間來解決這個問題

3.3 版本2-設定鎖的過期時間

public void lock() {
    while(true){
        String result = jedis.set(lockKey, "value", NOT_EXIST,SECONDS,30);
        if(OK.equals(result)){
            System.out.println(Thread.currentThread().getId()+"加鎖成功!");
            break;
        }
    }
}

類似的Redis命令如下

SET lockKey value NX EX 30

注：要保證設定過期時間和設定鎖具有原子性

這時又出現一個問題，問題出現的步驟如下

1. 客戶端A獲取鎖成功，過期時間30秒，
2. 客戶端A在某個操作上阻塞了50秒，
3. 30秒時間到了，鎖自動釋放了，
4. 客戶端B獲取到了對應同一個資源的鎖，
5. 客戶端A從阻塞中恢復過來，釋放掉了客戶端B持有的鎖，

示意圖如下

這時會有兩個問題

1. 過期時間如何保證大于業務執行時間?
2. 如何保證鎖不會被誤洗掉?

先來解決如何保證鎖不會被誤洗掉這個問題，

這個問題可以通過設定value為當前客戶端生成的一個隨機字串，且保證在足夠長的一段時間內在所有客戶端的所有獲取鎖的請求中都是唯一的，

3.4 版本3-設定鎖的value

抽象類RedisLock增加lockValue欄位，lockValue欄位的默認值為UUID隨機值假設當前執行緒ID，

public abstract class RedisLock implements Lock {

    //...
    protected String lockValue;

    public RedisLock(Jedis jedis,String lockKey) {
        this(jedis, lockKey, UUID.randomUUID().toString()+Thread.currentThread().getId());
    }

    public RedisLock(Jedis jedis, String lockKey, String lockValue) {
        this.jedis = jedis;
        this.lockKey = lockKey;
        this.lockValue = lockValue;
    }

    //...
}

加鎖代碼

public void lock() {
    while(true){
        String result = jedis.set(lockKey, lockValue, NOT_EXIST,SECONDS,30);
        if(OK.equals(result)){
            System.out.println(Thread.currentThread().getId()+"加鎖成功!");
            break;
        }
    }
}

解鎖代碼

public void unlock() {
    String lockValue = jedis.get(lockKey);
    if (lockValue.equals(lockValue)){
        jedis.del(lockKey);
    }
}

這時看看加鎖代碼，好像沒有什么問題啊，

再來看看解鎖的代碼，這里的解鎖操作包含三步操作：獲取值、判斷和洗掉鎖，這時你有沒有想到在多執行緒環境下的i++操作?

3.4.1 i++問題

i++操作也可分為三個步驟：讀i的值，進行i+1，設定i的值，

如果兩個執行緒同時對i進行i++操作，會出現如下情況

1. i設定值為0
2. 執行緒A讀到i的值為0
3. 執行緒B也讀到i的值為0
4. 執行緒A執行了+1操作，將結果值1寫入到記憶體
5. 執行緒B執行了+1操作，將結果值1寫入到記憶體
6. 此時i進行了兩次i++操作，但是結果卻為1

在多執行緒環境下有什么方式可以避免這類情況發生?

解決方式有很多種，例如用AtomicInteger、CAS、synchronized等等，

這些解決方式的目的都是要確保i++ 操作的原子性，那么回過頭來看看解鎖，同理我們也是要確保解鎖的原子性，我們可以利用Redis的lua腳本來實作解鎖操作的原子性，

3.5 版本4-具有原子性的釋放鎖

3.6 版本5-確保過期時間大于業務執行時間

3.7 測驗

累了，不想寫了，需要完整版的朋友點一點下方鏈接自己領取

鏈接：點這里！！！ 799215493 暗號：CSDN

4 集群Redis的分布式鎖

在Redis的分布式環境中，Redis 的作者提供了RedLock 的演算法來實作一個分布式鎖，

4.1 加鎖

RedLock演算法加鎖步驟如下

1. 獲取當前Unix時間，以毫秒為單位，
2. 依次嘗試從N個實體，使用相同的key和隨機值獲取鎖，在步驟2，當向Redis設定鎖時,客戶端應該設定一個網路連接和回應超時時間，這個超時時間應該小于鎖的失效時間，例如你的鎖自動失效時間為10秒，則超時時間應該在5-50毫秒之間，這樣可以避免服務器端Redis已經掛掉的情況下，客戶端還在死死地等待回應結果，如果服務器端沒有在規定時間內回應，客戶端應該盡快嘗試另外一個Redis實體，
3. 客戶端使用當前時間減去開始獲取鎖時間（步驟1記錄的時間）就得到獲取鎖使用的時間，當且僅當從大多數（這里是3個節點）的Redis節點都取到鎖，并且使用的時間小于鎖失效時間時，鎖才算獲取成功，
4. 如果取到了鎖，key的真正有效時間等于有效時間減去獲取鎖所使用的時間（步驟3計算的結果），
5. 如果因為某些原因，獲取鎖失敗（沒有在至少N/2+1個Redis實體取到鎖或者取鎖時間已經超過了有效時間），客戶端應該在所有的Redis實體上進行解鎖（即便某些Redis實體根本就沒有加鎖成功），

4.2 解鎖

向所有的Redis實體發送釋放鎖命令即可，不用關心之前有沒有從Redis實體成功獲取到鎖.

總結

我這里準備了一線大廠面試資料和我原創的超硬核PDF技術檔案，以及我為大家精心準備的多套簡歷模板（不斷更新中），希望大家都能找到心儀的作業！

有需要的朋友可以點一點下方鏈接免費領取

鏈接：點這里！！！ 799215493 暗號：CSDN