利用Redis實現分佈式鎖

寫在最前面

我在以前總結冪等性的時候，寫過一種分佈式鎖的實現，惋惜當時沒有真正應用過，着實的心虛啊。正好這段時間對這部分實踐了一下，也算是對以前填坑了。

分佈式鎖按照網上的結論，大體分爲三種：一、數據庫樂觀鎖； 二、基於Redis的分佈式鎖；3.、基於ZooKeeper的分佈式鎖；

關於樂觀鎖的實現其實在以前已經講的很清楚了，有興趣的移步：使用mysql樂觀鎖解決併發問題 。今天先簡單總結下redis的實現方法，後面詳細研究過ZooKeeper的實現原理後再具體說說ZooKeeper的實現。

爲何須要分佈式鎖？

在傳統單體應用單機部署的狀況下，可使用Java併發相關的鎖，如ReentrantLcok或synchronized進行互斥控制。可是，隨着業務發展的須要，原單體單機部署的系統，漸漸的被部署在多機器多JVM上同時提供服務，這使得原單機部署狀況下的併發控制鎖策略失效了，爲了解決這個問題就須要一種跨JVM的互斥機制來控制共享資源的訪問，這就是分佈式鎖要解決的問題。

分佈式鎖的實現條件

一、互斥性，和單體應用同樣，要保證任意時刻，只能有一個客戶端持有鎖

二、可靠性，要保證系統的穩定性，不能產生死鎖

三、一致性，要保證鎖只能由加鎖人解鎖，不能產生A的加鎖被B用戶解鎖的狀況

Redis分佈式鎖的實現

Redis實現分佈式鎖不一樣的人可能有不一樣的實現邏輯，可是核心就是下面三個方法。

SETNX
SETNX key val
當且僅當key不存在時，set一個key爲val的字符串，返回1；若key存在，則什麼都不作，返回0。
Expire
expire key timeout
爲key設置一個超時時間，單位爲second，超過這個時間鎖會自動釋放，避免死鎖。
Delete
delete key
刪除key

獲取鎖

首先講一個目前網上應用最多的一種實現

實現思路：

1.獲取鎖的時候，使用setnx加鎖，並使用expire命令爲鎖添加一個超時時間，超過該時間則自動釋放鎖以避免產生死鎖，鎖的value值爲一個隨機生成的UUID，經過此在釋放鎖的時候進行判斷。

2.獲取鎖的時候還設置一個獲取的超時時間，若超過這個時間則放棄獲取鎖。

3.釋放鎖的時候，經過UUID判斷是否是該鎖，如果該鎖，則執行delete進行鎖釋放。

public String getRedisLock(Jedis jedis, String lockKey, Long acquireTimeout, Long timeOut) {
        try {
            // 定義 redis 對應key 的value值(uuid) 做用 釋放鎖 隨機生成value,根據項目狀況修改
            String identifierValue = UUID.randomUUID().toString();
            // 定義在獲取鎖以後的超時時間
            int expireLock = (int) (timeOut / 1000);// 以秒爲單位
            // 定義在獲取鎖以前的超時時間
            //使用循環機制 若是沒有獲取到鎖，要在規定acquireTimeout時間 保證重複進行嘗試獲取鎖
            // 使用循環方式重試的獲取鎖
            Long endTime = System.currentTimeMillis() + acquireTimeout;
            while (System.currentTimeMillis() < endTime) {
                // 獲取鎖
                // 使用setnx命令插入對應的redislockKey ，若是返回爲1 成功獲取鎖
                if (jedis.setnx(lockKey, identifierValue) == 1) {
                    // 設置對應key的有效期
                    jedis.expire(lockKey, expireLock);
                    return identifierValue;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        } 
        return null;
    }複製代碼

這種實現方法也是目前應用最多的實現，我一直覺得這確實是正確的。然而因爲這是兩條Redis命令，不具備原子性，若是程序在執行完setnx()以後忽然崩潰，致使鎖沒有設置過時時間。那麼仍是會發生死鎖的狀況。網上之因此有人這樣實現，是由於低版本的jedis並不支持多參數的set()方法。

固然這種狀況Jedis的設計者也顯然想到了，新版的Jedis能夠同時set多個參數，具體實現以下：

實現思路：

基本上和原來的邏輯相似，只是將setnx和expire的操做合併爲一步，改成使用新的set多參的方法。

set(final String key, final String value, final String nxxx, final String expx,final long time)

key和value天然不用多說。nxxx參數只能夠傳String 類型的NX（僅在不存在的狀況下設置）和XX（和普通的set操做同樣會作更新操做）兩種。

expx是指到期時間單位，可傳參數爲EX （秒）和 PX （毫秒）,time就是具體的過時時間了，單位爲前面expx所指定的。

而後咱們對上面的代碼進行改造以下：

/**
     * @param acquireTimeout
     *            在獲取鎖以前的超時時間
     * @param timeOut
     *            在獲取鎖以後的超時時間
     */
    public String getRedisLock(Jedis jedis, String lockKey, Long acquireTimeout, Long timeOut) {
        try {
            // 定義 redis 對應key 的value值(uuid) 做用 釋放鎖 隨機生成value,根據項目狀況修改
            String identifierValue = UUID.randomUUID().toString();
            // 定義在獲取鎖以前的超時時間
            //使用循環機制 若是沒有獲取到鎖，要在規定acquireTimeout時間 保證重複進行嘗試獲取鎖
            // 使用循環方式重試的獲取鎖
            Long endTime = System.currentTimeMillis() + acquireTimeout;
            while (System.currentTimeMillis() < endTime) {
                // 獲取鎖
                // set使用NX參數的方式就等同於 setnx()方法，成功返回OK.PX以毫秒爲單位
                if ("OK".equals(jedis.set(lockKey, lockKey, "NX", "PX", timeOut))) {
                    return identifierValue;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        } 
        return null;
    }複製代碼

好了，獲取鎖的操做基本上就上面這些，有同窗可能要問，爲何不直接返回一個Boolean型的true或false呢？

正如我前面所說的，要保證解鎖的一致性，因此就須要經過value值來保證解鎖人就是加鎖人，而不能直接返回true或false了。

下面在說下解鎖的過程。

釋放鎖

仍是先舉一個錯誤的例子：

實現思路：

釋放鎖的時候，經過傳入key和加鎖時返回的value值，判斷傳入的value是否和key從redis中取出的相等。相等則證實解鎖人就是加鎖人，執行delete釋放鎖的操做。

// 釋放redis鎖
    public void unRedisLock(Jedis jedis, String lockKey, String identifierValue) {
        try {
            // 若是該鎖的id 等於identifierValue 是同一把鎖狀況才能夠刪除
            if (jedis.get(lockKey).equals(identifierValue)) {
                jedis.del(lockKey);
            }
        } catch (Exception e){
            e.printStackTrace();
        }
    }複製代碼

看着好像沒啥問題哈。然而仔細想一想又總感受哪裏不對。

若是在執行jedis.del(lockKey)操做以前，恰好鎖的過時時間到了，而這個時候又有別的客戶端取到了鎖，咱們在此時執行刪除操做，不是又不符合一致性的要求了嗎。

而後咱們修改成下述方案：

修改後的代碼爲：

public void unRedisLock(Jedis jedis, String lockKey, String identifierValue) {
        try {
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Long result = (Long) jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(identifierValue));
            //0釋放鎖失敗。1釋放成功
            if (1 == result) {
                //若是你想返回刪除成功仍是失敗，能夠在這裏返回
                System.out.println(result+"釋放鎖成功");
            } 
            if (0 == result){
                System.out.println(result+"釋放鎖失敗");
            }
        } catch (Exception e){
            e.printStackTrace();
        }
    }複製代碼

實現思路：

咱們將Lua代碼傳到jedis.eval()方法裏，並使參數KEYS[1]賦值爲lockKey，ARGV[1]賦值爲identifierValue。eval()方法是將Lua代碼交給Redis服務端執行。

那麼這段Lua代碼的功能是什麼呢？其實很簡單，首先獲取鎖對應的value值，檢查是否與identifierValue相等，若是相等則刪除鎖（解鎖）。那麼爲何要使用Lua語言來實現呢？由於要確保上述操做是原子性的。

那麼爲何執行eval()方法能夠確保原子性？源於Redis的特性，由於Redis是單線程，在eval命令執行Lua代碼的時候，Lua代碼將被當成一個命令去執行，而且直到eval命令執行完成，Redis纔會執行其餘命令。

若是想免費學習Java工程化、高性能及分佈式、深刻淺出。微服務、Spring，MyBatis，Netty源碼分析的朋友能夠加個人Java進階羣：478030634，羣裏有阿里大牛直播講解技術，以及Java大型互聯網技術的視頻免費分享給你們。

總結

本文對Redis實現分佈式鎖作了比較詳細的總結。我我的也對上述代碼作了實踐檢驗。其實我在使用時，一直用的錯誤的案例。直到看到園友Ruthless的一篇文章才曉得稀疏日常的寫法居然漏洞百出。下一篇準備再研究研究ZooKeeper的實現。