Redis分布式鎖怎么實現及應用場景是什么

本篇內容介紹了“Redis分布式鎖怎么實現及應用場景是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

引言

鎖是開發過程中十分常見的工具，你一定不陌生，悲觀鎖，樂觀鎖，排它鎖，公平鎖，非公平鎖等等，很多概念，如果你對java里的鎖還不了解，可以參考這一篇：不可不說的Java“鎖”事，這一篇寫的很全面了，但是對于初學者，知道這些鎖的概念，由于缺乏實際工作經驗，可能并不了解鎖的實際使用場景，Java中可以通過Volatile、Synchronized、ReentrantLock 三個關鍵字來實現線程的安全，這部分知識在第一輪基礎面試里一定會問（要熟練掌握哦）。

在分布式系統中Java這些鎖技術是無法同時鎖住兩臺機器上的代碼，所以要通過分布式鎖來實現，熟練使用分布式鎖也是大廠開發必會的技能。

1、面試官：

你有遇到需要使用分布式鎖的場景嗎？

問題分析：這個問題主要作為引子，先要了解什么場景下需要使用分布式鎖，分布式鎖要解決什么問題，在此前提下有助于你更好的理解分布式鎖的實現原理。

使用分布式鎖的場景一般需要滿足以下場景：

• 系統是一個分布式系統，java的鎖已經鎖不住了。

• 操作共享資源，比如庫里唯一的用戶數據。

• 同步訪問，即多個進程同時操作共享資源。

答：說一個我在項目中使用分布式鎖場景的例子：

消費積分在很多系統里都有，信用卡，電商網站，通過積分換禮品等，這里“消費積分”這個操作是需要使用鎖的典型場景。

事件A：

 以積分兌換禮品為例來講，完整的積分消費過程簡單分成3步：

A1：用戶選中商品，發起兌換提交訂單。

A2：系統讀取用戶剩余積分：判斷用戶當前積分是否充足。

A3：扣掉用戶積分。

事件B： 

系統給用戶發放積分也簡單分成3步：

B1：計算用戶當天應得積分

B2：讀取用戶原有積分

B3：在原有積分上增加本次應得積分

那么問題來了，如果用戶消費積分和用戶累加積分同時發生（同時用戶積分進行操作）會怎樣？

假設：用戶在消費積分的同時恰好離線任務在計算積分給用戶發放積分（如根據用戶當天的消費額），這兩件事同時進行，下面的邏輯有點繞，耐心理解。

用戶U有1000積分（記錄用戶積分的數據可以理解為共享資源），本次兌換要消耗掉999積分。

不加鎖的情況：事件A程序在執行到第2步讀積分時，A:2操作讀到的結果是1000分，判斷剩余積分夠本次兌換，緊接著要執行第3步A:3操作扣積分（1000 - 999 = 1），正常結果應該是用戶還是1分。但是這個時候事件B也在執行，本次要給用戶U發放100積分，兩個線程同時進行（同步訪問），不加鎖的情況，就會有下面這種可能，A:2 -> B:2 -> A:3 -> B:3 ,在A:3尚未完成前（扣積分，1000 - 999），用戶U總積分被事件B的線程讀取了，最后用戶U的總積分變成了1100分，還白白兌換了一個999積分的禮物，這顯然不符合預期結果。

有人說怎么可能這么巧同時操作用戶積分，cpu那么快，只要用戶足夠多，并發量足夠大，墨菲定律遲早生效，出現上述bug只是時間問題，還有可能被黑產行業卡住這個bug瘋狂薅羊毛，這個時候作為開發人員要解決這個隱患就必須了解鎖的使用。

（寫代碼是一項嚴謹的事兒！）

Java本身提供了兩種內置的鎖的實現，一種是由JVM實現的synchronized 和 JDK 提供的 Lock，以及很多原子操作類都是線程安全的，當你的應用是單機或者說單進程應用時，可以使用這兩種鎖來實現鎖。

但是當下互聯網公司的系統幾乎都是分布式的，這個時候Java自帶的 synchronized 或 Lock 已經無法滿足分布式環境下鎖的要求了，因為代碼會部署在多臺機器上，為了解決這個問題，分布式鎖應運而生，分布式鎖的特點是多進程，多個物理機器上無法共享內存，常見的解決辦法是基于內存層的干涉，落地方案就是基于Redis的分布式鎖 or ZooKeeper分布式鎖。

（我分析的不能更詳細了，面試官再不滿意？）

2、面試官：

Redis分布式鎖實現方法

問題分析：目前分布式鎖的實現方式主要有兩種，1.基于Redis Cluster模式。2.基于Zookeeper 集群模式。

優先掌握這兩種，應付面試基本沒問題了。

答：

1、基于Redis的分布式鎖

方法一：使用setnx命令加鎖

public static void wrongGetLock1(Jedis jedis, String lockKey, String requestId, int expireTime) {
		// 第一步：加鎖
    Long result = jedis.setnx(lockKey, requestId);
    if (result == 1) {
        // 第二步：設置過期時間
        jedis.expire(lockKey, expireTime);
    }
}

代碼解釋：

setnx命令，意思就是 set if not exist，如果lockKey不存在，把key存入Redis，保存成功后如果result返回1，表示設置成功，如果非1，表示失敗，別的線程已經設置過了。

expire()，設置過期時間，防止死鎖，假設，如果一個鎖set后，一直不刪掉，那這個鎖相當于一直存在，產生死鎖。

（講到這里，我還要和面試官強調一個“但是”）

思考，我上面的方法哪里與缺陷？繼續給面試官解釋…

加鎖總共分兩步，第一步jedis.setnx，第二步jedis.expire設置過期時間，setnx與expire不是一個原子操作，如果程序執行完第一步后異常了，第二步jedis.expire(lockKey, expireTime)沒有得到執行，相當于這個鎖沒有過期時間，有產生死鎖的可能。正對這個問題如何改進？

改進：

public class RedisLockDemo {
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    /**
     * 獲取分布式鎖
     * @param jedis Redis客戶端
     * @param lockKey 鎖
     * @param requestId 請求標識
     * @param expireTime 超期時間
     * @return 是否獲取成功
     */
    public static boolean getLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
				// 兩步合二為一，一行代碼加鎖并設置 + 過期時間。
        if (1 == jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime)) {
            return true;//加鎖成功
        }
        return false;//加鎖失敗
    }
}

代碼解釋：

將加鎖和設置過期時間合二為一，一行代碼搞定，原子操作。

(沒等面試官開口追問，面試官很滿意了)

3、面試官： 那解鎖操作呢？

答：

釋放鎖就是刪除key

使用del命令解鎖

public static void unLock(Jedis jedis, String lockKey, String requestId) {
    // 第一步： 使用 requestId 判斷加鎖與解鎖是不是同一個客戶端
    if (requestId.equals(jedis.get(lockKey))) {
        // 第二步： 若在此時，這把鎖突然不是這個客戶端的，則會誤解鎖
        jedis.del(lockKey);
    }
}

代碼解釋： 通過 requestId 判斷加鎖與解鎖是不是同一個客戶端和 jedis.del(lockKey) 兩步不是原子操作，理論上會出現在執行完第一步if判斷操作后鎖其實已經過期，并且被其它線程獲取，這是時候在執行jedis.del(lockKey)操作，相當于把別人的鎖釋放了，這是不合理的。當然，這是非常極端的情況，如果unLock方法里第一步和第二步沒有其它業務操作，把上面的代碼扔到線上，可能也不會真的出現問題，原因第一是業務并發量不高，根本不會暴露這個缺陷，那么問題還不大。

但是寫代碼是嚴謹的工作，能完美則必須完美。針對上述代碼中的問題，提出改進。

代碼改進：

public class RedisTool {
    private static final Long RELEASE_SUCCESS = 1L;
    /**
     * 釋放分布式鎖
     * @param jedis Redis客戶端
     * @param lockKey 鎖
     * @param requestId 請求標識
     * @return 是否釋放成功
     */
    public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;
    }
}

代碼解釋：

通過 jedis 客戶端的 eval 方法和 script 腳本一行代碼搞定，解決方法一中的原子問題。

3、面試官：

基于 ZooKeeper 的分布式鎖實現原理

答：還是積分消費與積分累加的例子：事件A和事件B同時需要進行對積分的修改操作，兩臺機器同時進行，正確的業務邏輯上讓一臺機器先執行完后另外一個機器再執行，要么事件A先執行，要么事件B先執行，這樣才能保證不會出現A:2 -> B:2 -> A:3 -> B:3這種積分越花越多的情況（想到這種bug一旦上線，老板要生氣了，我可能要哭了）。

怎么辦？使用 zookeeper 分布式鎖。

一個機器接收到了請求之后，先獲取 zookeeper 上的一把分布式鎖（zk會創建一個 znode），執行操作；然后另外一個機器也嘗試去創建那個 znode，結果發現自己創建不了，因為被別人創建了，那只能等待，等第一個機器執行完了方可拿到鎖。

使用 ZooKeeper 的順序節點特性，假如我們在/lock/目錄下創建3個節點，ZK集群會按照發起創建的順序來創建節點，節點分為/lock/0000000001、/lock/0000000002、/lock/0000000003，最后一位數是依次遞增的，節點名由zk來完成。

ZK中還有一種名為臨時節點的節點，臨時節點由某個客戶端創建，當客戶端與ZK集群斷開連接，則該節點自動被刪除。EPHEMERAL_SEQUENTIAL為臨時順序節點。

根據ZK中節點是否存在，可以作為分布式鎖的鎖狀態，以此來實現一個分布式鎖，下面是分布式鎖的基本邏輯：

• 客戶端調用create()方法創建名為“/dlm-locks/lockname/lock-”的臨時順序節點。

• 客戶端調用getChildren(“lockname”)方法來獲取所有已經創建的子節點。

• 客戶端獲取到所有子節點path之后，如果發現自己在步驟1中創建的節點是所有節點中序號最小的，就是看自己創建的序列號是否排第一，如果是第一，那么就認為這個客戶端獲得了鎖，在它前面沒有別的客戶端拿到鎖。

• 如果創建的節點不是所有節點中需要最小的，那么則監視比自己創建節點的序列號小的最大的節點，進入等待。直到下次監視的子節點變更的時候，再進行子節點的獲取，判斷是否獲取鎖。

釋放鎖的過程相對比較簡單，就是刪除自己創建的那個子節點即可，不過也仍需要考慮刪除節點失敗等異常情況。

額外補充

分布式鎖還可以從數據庫下手解決問題

方法一：

利用 Mysql 的鎖表，創建一張表，設置一個 UNIQUE KEY 這個 KEY 就是要鎖的 KEY，所以同一個 KEY 在mysql表里只能插入一次了，這樣對鎖的競爭就交給了數據庫，處理同一個 KEY 數據庫保證了只有一個節點能插入成功，其他節點都會插入失敗。

這樣 lock 和 unlock 的思路就很簡單了，偽代碼：

def lock ：
    exec sql: insert into locked—table (xxx) values (xxx)
    if result == true :
        return true
    else :
        return false
def unlock ：
    exec sql: delete from lockedOrder where order_id='order_id'

方法二：

使用流水號+時間戳做冪等操作，可以看作是一個不會釋放的鎖。

“Redis分布式鎖怎么實現及應用場景是什么”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！