集羣環境下Redis分佈式鎖的正確姿式

RedLock算法的實現思路

antirez提出的redlock算法實現思路大概是這樣的。

客戶端按照下面的步驟來獲取鎖：

1. 獲取當前時間的毫秒數T1。
2. 按順序依次向N個Redis節點執行獲取鎖的操做。這個獲取鎖的操做和上一篇中基於單Redis節點獲取鎖的過程相同。包括惟一UUID做爲Value以及鎖的過時時間(expireTime)。爲了保證在某個在某個Redis節點不可用的時候算法可以繼續運行，這個獲取鎖的操做還須要一個超時時間。它應該遠小於鎖的過時時間。客戶端向某個Redis節點獲取鎖失敗後，應當即嘗試下一個Redis節點。這裏失敗包括Redis節點不可用或者該Redis節點上的鎖已經被其餘客戶端持有。
3. 計算整個獲取鎖過程的總耗時。即當前時間減去第一步記錄的時間。計算公司爲T2=now()- T1。若是客戶端從大多數Redis節點(>N/2 +1)成功獲取到鎖。而且獲取鎖總共消耗的時間小於鎖的過時時間（即T2<expireTime）。則認爲客戶端獲取鎖成功，不然，認爲獲取鎖失敗
4. 若是獲取鎖成功，須要從新計算鎖的過時時間。它等於最初鎖的有效時間減去第三步計算出來獲取鎖消耗的時間，即expireTime - T2
5. 若是最終獲取鎖失敗，那麼客戶端當即向全部Redis系欸但發起釋放鎖的操做。（和上一篇釋放鎖的邏輯同樣）

雖說RedLock算法能夠解決單點Redis分佈式鎖的安全性問題，但若是集羣中有節點發生崩潰重啓，仍是會鎖的安全性有影響的。具體出現問題的場景以下：

假設一共有5個Redis節點：A, B, C, D, E。設想發生了以下的事件序列：

1. 客戶端1成功鎖住了A, B, C，獲取鎖成功（但D和E沒有鎖住）
2. 節點C崩潰重啓了，但客戶端1在C上加的鎖沒有持久化下來，丟失了
3. 節點C重啓後，客戶端2鎖住了C, D, E，獲取鎖成功

這樣，客戶端1和客戶端2同時得到了鎖（針對同一資源）。針對這樣場景，解決方式也很簡單，也就是讓Redis崩潰後延遲重啓，而且這個延遲時間大於鎖的過時時間就好。這樣等節點重啓後，全部節點上的鎖都已經失效了。也不存在以上出現2個客戶端獲取同一個資源的狀況了。 

RedLock安全性和穩定性好，但要說徹底沒有問題不是。例如，若是客戶端獲取鎖成功後，若是訪問共享資源操做執行時間過長，致使鎖過時了，後續客戶端獲取鎖成功了，這樣在同一個時刻又出現了2個客戶端得到了鎖的狀況。因此針對分佈式鎖的應用的時候須要多測試。服務器臺數越多，出現不可預期的狀況也越多。若是客戶端獲取鎖以後，在上面第三步發生了GC得狀況致使GC完成後，鎖失效了，這樣同時也使得同一時間有2個客戶端得到了鎖。若是系統對共享資源有很是嚴格要求得狀況下，仍是建議須要作數據庫鎖得得方案來補充。如飛機票或火車票座位得狀況。對於一些搶購獲取，針對偶爾出現超賣，後續能夠人爲溝通置換得方式採用分佈式鎖得方式沒什麼問題。由於能夠絕大部分保證分佈式鎖的安全性。

分佈式場景下基於Redis實現分佈式鎖的正確姿式

目前redisson包已經有對redlock算法封裝，接下來就具體看看使用redisson包來實現分佈式鎖的正確姿式。

具體實現代碼以下代碼所示：

public interface DistributedLock {
    /**
     * 獲取鎖
     * @return 鎖標識
     */
    String acquire();

    /**
     * 釋放鎖
     * @param indentifier
     * @return
     */
    boolean release(String indentifier);
}

public class RedisDistributedRedLock implements DistributedLock {

    /**
     * redis 客戶端
     */
    private RedissonClient redissonClient;

    /**
     * 分佈式鎖的鍵值
     */
    private String lockKey;

    private RLock redLock;

    /**
     * 鎖的有效時間 10s
     */
    int expireTime = 10 * 1000;

    /**
     * 獲取鎖的超時時間
     */
    int acquireTimeout  = 500;

    public RedisDistributedRedLock(RedissonClient redissonClient, String lockKey) {
        this.redissonClient = redissonClient;
        this.lockKey = lockKey;
    }

    @Override
    public String acquire() {
        redLock = redissonClient.getLock(lockKey);
        boolean isLock;
        try{
            isLock = redLock.tryLock(acquireTimeout, expireTime, TimeUnit.MILLISECONDS);
            if(isLock){
                System.out.println(Thread.currentThread().getName() + " " + lockKey + "得到了鎖");
                return null;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public boolean release(String indentifier) {
        if(null != redLock){
            redLock.unlock();
            return true;
        }

        return false;
    }
}
複製代碼

因爲RedLock是針對主從和集羣場景準備。上面代碼採用哨兵模式。因此要讓上面代碼運行起來，須要先本地搭建Redis哨兵模式。

具體測試代碼以下所示：

public class RedisDistributedRedLockTest {
    static int n = 5;
    public static void secskill() {
        if(n <= 0) {
            System.out.println("搶購完成");
            return;
        }

        System.out.println(--n);
    }
    public static void main(String[] args) {

        Config config = new Config();
        //支持單機，主從，哨兵，集羣等模式
        //此爲哨兵模式
        config.useSentinelServers()
                .setMasterName("mymaster")
                .addSentinelAddress("127.0.0.1:26369","127.0.0.1:26379","127.0.0.1:26389")
                .setDatabase(0);
        Runnable runnable = () -> {
            RedisDistributedRedLock redisDistributedRedLock = null;
            RedissonClient redissonClient = null;
            try {
                redissonClient = Redisson.create(config);
                redisDistributedRedLock = new RedisDistributedRedLock(redissonClient, "stock_lock");
                redisDistributedRedLock.acquire();
                secskill();
                System.out.println(Thread.currentThread().getName() + "正在運行");
            } finally {
                if (redisDistributedRedLock != null) {
                    redisDistributedRedLock.release(null);
                }

                redissonClient.shutdown();
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
複製代碼

總結

到此，基於Redis實現分佈式鎖的就告一段落了，因爲分佈式鎖的實現方式主要有：數據庫鎖的方式、基於Redis實現和基於Zookeeper實現。