【分散式緩存系列】集群環境下Redis分散式鎖的正確姿勢

一、前言在上一篇文章中，已經介紹了基於Redis實現分散式鎖的正確姿勢，但是上篇文章存在一定的缺陷——它加鎖只作用在一個Redis節點上，如果通過sentinel保證高可用，如果master節點由於某些原因發生了主從切換，那麼就會出現鎖丟失的情況：於是，客戶端1和客戶端2同時持有了同一個資源的鎖 ...

一、前言

　　在上一篇文章中，已經介紹了基於Redis實現分散式鎖的正確姿勢，但是上篇文章存在一定的缺陷——它加鎖只作用在一個Redis節點上，如果通過sentinel保證高可用，如果master節點由於某些原因發生了主從切換，那麼就會出現鎖丟失的情況：

客戶端1在Redis的master節點上拿到了鎖
Master宕機了，存儲鎖的key還沒有來得及同步到Slave上
master故障，發生故障轉移，slave節點升級為master節點
客戶端2從新的Master獲取到了對應同一個資源的鎖

　　於是，客戶端1和客戶端2同時持有了同一個資源的鎖。鎖的安全性被打破了。針對這個問題。Redis作者antirez提出了RedLock演算法來解決這個問題

二、RedLock演算法的實現思路

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

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

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

　　雖然說RedLock演算法可以解決單點Redis分散式鎖的安全性問題，但如果集群中有節點發生崩潰重啟，還是會鎖的安全性有影響的。具體出現問題的場景如下：

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

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

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

　　相比之下，RedLock安全性和穩定性都比前一篇文章中介紹的實現要好很多，但要說完全沒有問題不是。例如，如果客戶端獲取鎖成功後，如果訪問共用資源操作執行時間過長，導致鎖過期了，後續客戶端獲取鎖成功了，這樣在同一個時刻又出現了2個客戶端獲得了鎖的情況。所以針對分散式鎖的應用的時候需要多測試。伺服器台數越多，出現不可預期的情況也越多。如果客戶端獲取鎖之後，在上面第三步發生了GC得情況導致GC完成後，鎖失效了，這樣同時也使得同一時間有2個客戶端獲得了鎖。如果系統對共用資源有非常嚴格要求得情況下，還是建議需要做資料庫鎖得得方案來補充。如飛機票或火車票座位得情況。對於一些搶購獲取，針對偶爾出現超賣，後續可以人為溝通置換得方式採用分散式鎖得方式沒什麼問題。因為可以絕大部分保證分散式鎖的安全性。

三、分散式場景下基於Redis實現分散式鎖的正確姿勢

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

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

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

    /**
     * 釋放鎖
     * @author zhi.li
     * @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哨兵模式。本人的環境是Windows,具體Windows 哨兵環境搭建參考文章：redis sentinel部署(Windows下實現)。

　　具體測試代碼如下所示：

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();
        }
    }