10 分鐘精通 Redis 分佈式鎖中的各種門道

作者 | 一小頁責編 | 王曉曼

出品 | CSDN 博客

道阻且長，行則將至。請相信我，你一定會更優秀！

本文我們主要聊 Redis 實現分佈式鎖，別的不聊。先來三個問題熱熱身：

一個 setnx 就行了？value 沒意義？還有人認爲 incr 也可以？
再加個超時時間就行了？
你寫的分佈式鎖，你確認你敢投產嗎？

「 Redis 做分佈式鎖，沒那麼簡單。調整好心態，保證你有收穫」

1、爲什麼要有分佈式鎖？

2、先捋脈絡，再想風險，最後再寫代碼

3、一步一步，看透 Redis 分佈式鎖中的門道

3-1、放在 finally{} 塊中就行了嗎？
3-2、鎖的超時時間該怎麼計算？
3-3、加個超時時間就行了嗎？
3-4、這個密語 value 設置成什麼呢？
3-5、繼續，現在把思維先跳出來，想想？可重入怎麼搞？
3-6、能不侵入業務代碼嗎？
3-7、Thread-Id 真能行嗎？
3-8、APP_ID + ThreadId 還是 UUID 好呢？
3-9、鎖重入，就這麼簡單？
3-10、重入鎖的方法中直接執行 unlock？這麼大膽！
3-11、終於見到希望？再來一招！
3-12、搞了半天，鎖還是崩潰了？
3-13、別急，還有。鎖在我手裏，我掛了，這...

4、風險！主從部署引來的問題

爲什麼要有分佈式鎖？

JUC 提供的鎖機制，可以保證在同一個 JVM 進程中同一時刻只有一個線程執行操作邏輯；多服務多節點的情況下，就意味着有多個 JVM 進程，要做到這樣，就需要有一箇中間人；

分佈式鎖就是用來保證在同一時刻，僅有一個 JVM 進程中的一個線程在執行操作邏輯；換句話說，JUC 的鎖和分佈式鎖都是一種保護系統資源的措施。儘可能將併發帶來的不確定性轉換爲同步的確定性；

先捋脈絡，再想風險，最後再寫代碼

當我們設計一個東西的時候，很多同學腦子裏想到的第一件事就是代碼，代碼。聽我說，你一定要先思考，要做一根能思想的葦草。代碼是死的，三思而後行。

所以，一定要先在腦子裏想，這把鎖，我要用它幹什麼，它要保證什麼，有沒有什麼意外情況，會存在什麼風險。先全局看一下，別一下子鑽到裏邊。想完了之後，然後一定要落地，絕對不可以紙上談兵，自己一定要把代碼寫出來，自己去測試，去解決問題，看到底行不行。只有寫出來，你才能驗證你的想法。“實踐是檢驗真理的唯一標準”。

爲了保證文章的易讀性，接下來，我將採用理論 + 代碼的形式，從整體到部分，從宏觀到微觀，帶你全面看透 Redis 分佈式鎖。

一步一步，看透 Redis 分佈式鎖中的門道

我們一起捋一下，很多線程去上鎖，誰鎖成功誰就有權利執行操作邏輯，其他線程要麼直接走搶鎖失敗的邏輯，要麼自旋嘗試搶鎖；

比方說 A 線程競爭到了鎖，開始執行操作邏輯（我的代碼邏輯演示中，使用 Jedis 客戶端爲例）；

public static void doSomething() {
    // RedisLock是我封裝的一個類，後面會講到
    RedisLock redisLock = new RedisLock(jedis); // 創建jedis實例的代碼省略，不是重點
    try {
        redisLock.lock(); // 上鎖
        // 處理業務
        System.out.println(Thread.currentThread().getName() + " 線程處理業務邏輯中...");
        Thread.sleep(2000);
        System.out.println(Thread.currentThread().getName() + " 線程處理業務邏輯完畢");
        redisLock.unlock(); // 釋放鎖
    } catch (Exception e) {
        e.printStackTrace();
    }
}

正常情況下，A 線程執行完操作邏輯後，應該將鎖釋放。如果說執行過程中拋出異常，程序不再繼續走正常的釋放鎖流程，沒有釋放鎖怎麼辦？所以我們想到：

釋放鎖的流程一定要在 finally{} 塊中執行，當然，上鎖的流程一定要在 finally{} 對應的 try{} 塊中，否則 finally{} 就沒用了，如下：

public static void doSomething() {
    RedisLock redisLock = new RedisLock(jedis); // 創建jedis實例的代碼省略，不是重點
    try {
        redisLock.lock(); // 上鎖，必須在 try{}中
        // 處理業務
        System.out.println(Thread.currentThread().getName() + " 線程處理業務邏輯中...");
        Thread.sleep(2000);
        System.out.println(Thread.currentThread().getName() + " 線程處理業務邏輯完畢");
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        redisLock.unlock(); // 在finally{} 中釋放鎖
    }
}

3-1、放在 finally{} 塊中就行了嗎？

如果在執行 try{} 中邏輯的時候，程序出現了 System.exit(0); 或者 finally{} 中執行異常，比方說連接不上 redis-server 了；或者還未執行到 finally{} 的時候，JVM 進程掛掉了，服務宕機；這些情況都會導致沒有成功釋放鎖，別的線程一直拿不到鎖，怎麼辦？

如果我的系統因爲一個節點影響，別的節點也都無法正常提供服務了，那我的系統也太弱了。所以我們想到必須要將風險降低，可以給鎖設置一個超時時間，比方說 1 秒，即便發生了上邊的情況，那我的鎖也會在 1 秒之後自動釋放，其他線程就可以獲取到鎖，接班幹活了；

   public static final String lock_key = "haolin-lock";
     public void lock() {    
    while (!tryLock()) {
      try {
        Thread.sleep(50); // 在while中自旋，如果說讀者想設置一些自旋次數，等待最大時長等自己去擴展，不是此處的重點
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println("線程：" + threadName + "，佔鎖成功！★★★");
   }
    private boolean tryLock() {
    SetParams setParams = new SetParams();
    setParams.ex(1); // 超時時間1s
    setParams.nx();  // nx
    String response = jedis.set(lock_key, "", setParams); // 轉換爲redis命令就是：set haolin-key "" ex 1 nx
    return "OK".equals(response);
   }

注意，上鎖的時候，設置 key 和設置超時時間這兩個操作要是原子性的，要麼都執行，要麼都不執行。

Redis 原生支持：

// http://redis.io/commands/set.html
SET key value [EX seconds] [PX milliseconds] [NX|XX]

不要在代碼裏邊分兩次調用：

set k vexipre k time

這是錯誤的，如果第一個命令執行成功後，第二條命令由於各種原因沒有執行，就出問題了。

3-2、鎖的超時時間該怎麼計算？

我們剛纔假設的 1s 是怎麼計算的？這個時間該設多少合適呢？

聽我說，鎖中的業務邏輯的執行時間，不能瞎寫，一般是我們在測試環境進行多次測試，然後在壓測環境多輪壓測之後，比方說計算出平均的執行時間是 200ms，鎖的超時時間放大 3-5 倍，比如這裏我們設置爲 1s，爲啥要放大，因爲如果鎖的操作邏輯中有網絡 IO 操作，線上的網絡不會總一帆風順，我們要給網絡抖動留有緩衝時間。

這個時候有的同學有想法，那我設置的再大一些，給網絡足夠充裕的時間，我就設置 10s、1min 不是更安全嗎？請注意，不要鑽到這一個點裏邊，你要顧全大局，多大算大？越大越好？無窮大？那不等於不設置超時時間嗎？

同時，這個時間，你要想清楚，如果你設置 10s，果真發生了宕機，那意味着這 10s 中間，你的這個分佈式鎖的服務全部節點都是不可用的，這個和你的業務以及系統的可用性有掛鉤，你要去衡量，要慎重（後邊 3-13 會再詳細聊）。那如果一個節點宕機之後可以通知 redis-server 釋放鎖嗎？

注意，我是宕機，不可控力，斷電了兄弟，通知不了的。回頭一想，如果我是優雅停機呢，我不是 kill -9，也不是斷電，這樣似乎可以去做一些編碼去釋放鎖，你可以參考下 JVM 的鉤子、Dubbo 的優雅停機、或者 linux 進程級通信技術來做這件事情。

當然也可以手動停服務後，手動刪除掉 redis 中的鎖，讓他及時刪除掉，只要運維不打你。（大家都知道 zookeeper 的臨時節點特性被應用在很多場合，但本文不是聊 zookeeper，思路別跑題，迴歸）

3-3、加個超時時間就行了嗎？

繼續，如果說 A 線程在執行操作邏輯的過程中，別的線程直接進行了釋放鎖的操作，是不是就出問題了？

什麼？別的線程沒有獲得鎖卻直接執行了釋放鎖？？現在是 A 線程上的鎖，那肯定只能 A 線程釋放鎖呀！別的線程釋放鎖算怎麼回事？聯想 ReentrantLock 中的 isHeldByCurrentThread() 方法，所以我們想到，必須在鎖上加個標記，只有上鎖的線程 A 線程知道，相當於是一個密語，也就是說釋放鎖的時候，首先先把密語和鎖上的標記進行匹配，如果匹配不上，就沒有權利釋放鎖；

   private boolean tryLock() {
    SetParams setParams = new SetParams();
    setParams.ex(1); // 超時時間1s
    setParams.nx();  // nx
    String response = jedis.set(lock_key, "", setParams); // 轉換爲redis命令就是：set haolin-key "" ex 1 nx
    return "OK".equals(response);
  }
    // 別的線程直接調用釋放鎖操作，分佈式鎖崩潰！
   public void unlock() {
    jedis.del(encode(lock_key));
    System.out.println("線程：" + threadName + " 釋放鎖成功！☆☆☆");
  }
   private byte[] encode(String param) {
    return param.getBytes();
  }

3-4、這個密語 value 設置成什麼呢？

這是有門道的，跟着我的思路走，繼續。

很多同學說設置成一個 UUID 就行了，上鎖之前，在該線程代碼中生成一個 UUID，將這個作爲祕鑰，存在鎖鍵的 value 中，釋放鎖的時候，用這個進行校驗，因爲只有上鎖的線程知道這個祕鑰，別的線程是不知道的。這個可行嗎，當然可行。

   String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + 
        "then\n" + 
        "    return redis.call(\"del\", KEYS[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
    private boolean tryLock(String uuid) {
    SetParams setParams = new SetParams();
    setParams.ex(1); // 超時時間1s
    setParams.nx();  // nx
    String response = jedis.set(lock_key, uuid, setParams); // 轉換爲redis命令就是：set haolin-key "" ex 1 nx
    return "OK".equals(response);
  }
   public void unlock(String uuid) {
    List<byte[]> keys = Arrays.asList(encode(lock_key));
    List<byte[]> args = Arrays.asList(encode(uuid));
           // 使用lua腳本，保證原子性
    long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args);
    if (eval == 1) {
      System.out.println("線程：" + threadName + " 釋放鎖成功！☆☆☆");
    } else {
      System.out.println("線程：" + threadName + " 釋放鎖失敗！該線程未持有鎖！！！");
    }
  }
   private byte[] encode(String param) {
    return param.getBytes();
  }

爲什麼使用 lua 腳本？

保證原子性，因爲是兩個操作，如果分兩步那就是：

get k // 進行祕鑰 value的比對
del k // 比對成功後，刪除k

如果第一步比對成功後，第二步還沒來得及執行的時候，鎖到期，然後緊接着別的線程獲取到鎖，裏邊的 uuid 已經變了，也就是說持有鎖的線程已經不是該線程了，此時再執行第二步的刪除鎖操作，肯定是錯誤的了。

3-5、繼續，現在把思維先跳出來，想想？可重入怎麼搞？

作爲一把鎖，我們在使用 synchronized、ReentrantLock 的時候是不是有可重入性？

那咱們這把分佈式鎖該如何實現可重入呢？如果 A 線程的鎖方法邏輯中調用了 x() 方法，x() 方法中也需要獲取這把鎖，按照這個邏輯，x() 方法中的鎖應該重入進去即可，那是不是需要將剛纔生成的這個 UUID 祕鑰傳遞給 x() 方法？怎麼傳遞？參數？這就侵入業務代碼了。

3-6、能不侵入業務代碼嗎？

我們主要是想給上鎖的 A 線程設置一個只有它自己知道的祕鑰，把思路時鐘往回撥，想想：

線程本身的 id（Thread.currentThread().getId()）是不是就是一個唯一標識呢？我們把祕鑰 value 設置爲線程的 id 不就行了。

   String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + 
        "then\n" + 
        "    return redis.call(\"del\", KEYS[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
    String addLockLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + 
        "then\n" + 
        "    return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
     public void lock() {
             // 判斷是否可重入
    if (isHeldByCurrentThread()) {
      return;
    }
    while (!tryLock()) {
      try {
        Thread.sleep(50); // 自旋
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println("線程：" + threadName + "，佔鎖成功！★★★");
  }
   // 是否是當前線程佔有鎖，同時將超時時間重新設置，這個很重要，同樣也是原子操作
   private boolean isHeldByCurrentThread() {
    List<byte[]> keys = Arrays.asList(encode(lock_key));
    List<byte[]> args = Arrays.asList(encode(String.valueOf(threadId)), encode(String.valueOf(1)));
    long eval = (Long) jedis.eval(encode(addLockLife_lua), keys, args);
    return eval == 1;
  }
    private boolean tryLock(String uuid) {
    SetParams setParams = new SetParams();
    setParams.ex(1); // 超時時間1s
    setParams.nx();  // nx
    String response = jedis.set(lock_key, String.valueOf(threadId), setParams); // 轉換爲redis命令就是：set haolin-key xxx ex 1 nx
    return "OK".equals(response);
  }
   public void unlock(String uuid) {
    List<byte[]> keys = Arrays.asList(encode(lock_key));
    List<byte[]> args = Arrays.asList(encode(String.valueOf(threadId)));
           // 使用lua腳本，保證原子性
    long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args);
    if (eval == 1) {
      System.out.println("線程：" + threadName + " 釋放鎖成功！☆☆☆");
    } else {
      System.out.println("線程：" + threadName + " 釋放鎖失敗！該線程未持有鎖！！！");
    }
  }
   private byte[] encode(String param) {
    return param.getBytes();
  }

3-7、Thread-Id 真能行嗎？

不行。

想想，我們說一個 Thread 的 id 是唯一的，是在同一個 JVM 進程中，是在一個操作系統中，也就是在一個機器中。

而現實是，我們的部署是集羣部署，多個實例節點，那意味着會存在這樣一種情況，S1 機器上的線程上鎖成功，此時鎖中祕鑰 value 是線程 id=1，如果說同一時間 S2 機器中，正好線程 id=1 的線程嘗試獲得這把鎖，比對祕鑰發現成功，結果也重入了這把鎖，也開始執行邏輯，此時，我們的分佈式鎖崩潰！怎麼解決？

我們只需要在每個節點中維護不同的標識即可，怎麼維護呢？應用啓動的時候，使用 UUID 生成一個唯一標識 APP_ID，放在內存中（或者使用 zookeeper 去分配機器 id 等等）。此時，我們的祕鑰 value 這樣存即可：APP_ID+ThreadId

   // static變量，final修飾，加載在內存中，JVM進程生命週期中不變
   private static final String APP_ID = UUID.randomUUID().toString();
    String releaseLock_lua = "if redis.call(\"get\",KEYS[1]) == ARGV[1] \n" + 
        "then\n" + 
        "    return redis.call(\"del\", KEYS[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
    String addLockLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + 
        "then\n" + 
        "    return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
     public void lock() {
             // 判斷是否可重入
    if (isHeldByCurrentThread()) {
      return;
    }
    while (!tryLock()) {
      try {
        Thread.sleep(50); // 自旋
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println("線程：" + threadName + "，佔鎖成功！★★★");
  }
    // 是否是當前線程佔有鎖，同時將超時時間重新設置，這個很重要，同樣也是原子操作
   private boolean isHeldByCurrentThread() {
    List<byte[]> keys = Arrays.asList(encode(lock_key));
    List<byte[]> args = Arrays.asList(encode(APP_ID + String.valueOf(threadId)), encode(String.valueOf(1)));
    long eval = (Long) jedis.eval(encode(addLockLife_lua), keys, args);
    return eval == 1;
  }
    private boolean tryLock(String uuid) {
    SetParams setParams = new SetParams();
    setParams.ex(1); // 超時時間1s
    setParams.nx();  // nx
    String response = jedis.set(lock_key, APP_ID + String.valueOf(threadId), setParams); // 轉換爲redis命令就是：set haolin-key xxx ex 1 nx
    return "OK".equals(response);
  }
   public void unlock(String uuid) {
    List<byte[]> keys = Arrays.asList(encode(lock_key));
    List<byte[]> args = Arrays.asList(encode(APP_ID + String.valueOf(threadId)));
           // 使用lua腳本，保證原子性
    long eval = (Long) jedis.eval(encode(releaseLock_lua), keys, args);
    if (eval == 1) {
      System.out.println("線程：" + threadName + " 釋放鎖成功！☆☆☆");
    } else {
      System.out.println("線程：" + threadName + " 釋放鎖失敗！該線程未持有鎖！！！");
    }
  }
   private byte[] encode(String param) {
    return param.getBytes();
  }

3-8、APP_ID + ThreadId 還是 UUID 好呢？

是不是覺得有點意思了？

繼續聽我說，如果 A 線程執行邏輯中間開啓了一個子線程執行任務，這個子線程任務中也需要重入這把鎖，因爲子線程獲取到的線程 id 不一樣，導致重入失敗。那意味着需要將這個祕鑰繼續傳遞給子線程，JUC 中 InheritableThreadLocal 派上用場，但是感覺怪怪的，因爲線程間傳遞的是父線程的 id。

微服務中多服務間調用的話可以借用系統自身有的 traceId 作爲祕鑰即可。

「至於選擇哪種 value 的方式，根據實際的系統設計 + 業務場景，選擇最合適的即可，沒有最好，只有最合適。」

3-9、鎖重入，就這麼簡單？

注意，我們上邊的主要注意力在怎麼重入進去，而我們這是分佈式鎖，要考慮的事情還有很多，重入進去後，超時時間隨便設嗎？有門道嗎？

比方說 A 線程在鎖方法中調用了 x() 方法，而 x() 方法中也有獲取鎖的邏輯，如果 A 線程獲取鎖後，執行過程中，到 x() 方法時，這把鎖是要重入進去的，但是請注意，這把鎖的超時時間如果小於第一次上鎖的時間，比方說 A 線程設置的超時時間是 1s，在 100ms 的時候執行到 x() 方法中，而 x() 方法中設置的超時時間是 100ms，那麼意味着 100ms 之後鎖就釋放了，而這個時候我的 A 線程的主方法還沒有執行完呢！卻被重入鎖設置的時間搞壞了！這個怎麼搞？

如果說我在內存中設置一個這把鎖設置過的最大的超時時間，重入的時候判斷下傳進來的時間，我重入時 expire 的時候始終設置成最大的時間，而不是由重入鎖隨意降低鎖時間導致上一步的主鎖出現問題

放在內存中行嗎？我們上邊舉例中，調用的 x() 方法是在一個 JVM 中，如果是調用遠程的一個 RPC 服務呢（像這種調用的話就需要將祕鑰 value 通過 RpcContext 傳遞過去了）到另一個節點的服務中進行鎖重入，這個時間依然是要用當前設置過鎖的最大時間的，所以這個最大的時間要存在 redis 中而非 JVM 內存中

經過這一步的分析，我們的重入 lua 腳本就修改爲這樣了：

  ADD_LOCK_LIFE("if redis.call(\"get\", KEYS[1]) == ARGV[1]\n" +   // 判斷是否是鎖持有者
        "then\n" + 
        "    local thisLockMaxTimeKeepKey=KEYS[1] .. \":maxTime\"\n" +  // 記錄鎖最大時間的key是：鎖名字:maxTime
        "    local nowTime=tonumber(ARGV[2])\n" +  // 當前傳參進來的time
        "    local maxTime=redis.call(\"incr\", thisLockMaxTimeKeepKey)\n" + // 取出當前鎖設置的最大的超時時間，如果這個保持時間的key不存在返回的是字符串nil，這裏爲了lua腳本的易讀性，用incr操作，這樣讀出來的都是number類型的操作
        "    local bigerTime=maxTime\n" + // 臨時變量bigerTime=maxTime
        "    if nowTime>maxTime-1\n" +    // 如果傳參進來的時間>記錄的最大時間
        "    then\n" + 
        "        bigerTime=nowTime\n" + // 則更新bigerTime
        "        redis.call(\"set\", thisLockMaxTimeKeepKey, tostring(bigerTime))\n" + // 設置超時時間爲最大的time，是最安全的
        "    else \n" + 
        "        redis.call(\"decr\", thisLockMaxTimeKeepKey)\n" + // 當前傳參time<maxTime，將剛纔那次incr減回來
        "    end\n" + 
        "    return redis.call(\"expire\", KEYS[1], tostring(bigerTime))\n" + // 重新設置超時時間爲當前鎖過的最大的time
        "else\n" + 
        "    return 0\n" + 
        "end"),

其實，還有另外一種方案比較簡單，就是鎖的超時時間 = 第一次上鎖的時間 + 後面所有重入鎖的時間。也就是（expire = 主 ttl + 重入 exipre），這種方案是放大的思想，一放大就又有上邊提到過的一個問題：expire 太大怎麼辦，參考上邊。

3-10、重入鎖的方法中直接執行 unlock？這麼大膽！

A 線程執行一共需要 500ms，執行中需要調用 x() 方法，x() 方法中有一個重入鎖，執行用了 50ms，然後執行完後，x() 方法的 finally{} 塊中將鎖進行釋放。

爲啥能釋放掉？因爲祕鑰我有，匹配成功了我就直接釋放了。

這當然是有問題的，所以我們要通過鎖重入次數來進行釋放鎖時候的判斷，也就是說上鎖的時候需要多維護一個 key 來保存當前鎖的重入次數，如果執行釋放鎖時，先進行重入次數 -1，-1 後如果是 0，可以直接 del，如果 > 0，說明還有重入的鎖在，不能直接 del

目前爲止，算上上一步中設置最大超時時間的 key，加上這一步重入次數的 key，加上鎖本身的 key，已經有 3 個 key，需要注意的事情是，這三個 key 的超時時間是都要設置的！爲什麼？

假如說重入次數的 key 沒有設置超時時間，服務 A 節點中在一個 JVM 中重入了 5 次後，調用一次 RPC 服務，RPC 服務中同樣重入鎖，此時，鎖重入次數是 6，這個時候 A 服務宕機，就意味着無論怎樣，這把鎖不可能釋放了，這個分佈式鎖提供的完整能力，全線不可用了！

所以，這幾個 key 是要設置超時時間的！怎麼設置？我上一個鎖要維護這麼多 key 的超時時間？太複雜了吧，多則亂，則容易出問題。怎麼辦？

我們想一下，是不是最大超時時間的 key 和重入次數的 key，都附屬於鎖，它們都是鎖的屬性，如果鎖不在了，談它們就毫無意義，這個時候用什麼存儲呢？

redis 的 hash 數據結構，就可以做，key 是鎖，裏邊的 hashKey 分別是鎖的屬性， hashValue 是屬性值，超時時間只設置鎖本身 key 就可以了。這個時候，我們的鎖的數據結構就要改變一下了。我這兒就不寫這個了，大家意會之後可以嘗試去改一改，寫一寫。

3-11、終於見到希望？再來一招！

再一次把思路時鐘往回撥，回撥到設置超時時間那裏，我們預估鎖方法執行時間是 200ms，我們放大 5 倍後，設置超時時間是 1s，假想一下，如果生產環境中，鎖方法中的 IO 操作，極端情況下超時嚴重，比方說 IO 就消耗了 2s，那就意味着，在這次 IO 還沒有結束的時候，我這把鎖已經到期釋放掉了，就意味着別的線程趁虛而入，分佈式鎖崩潰！

3-12、搞了半天，鎖還是崩潰了？

跟着我的思路走，別放棄。

再一次把思維從現在的框框裏跳出來，想一想，我們要做的是一把分佈式鎖，想要的目的是同一時刻只有一個線程持有鎖，作爲服務而言，這個鎖現在不管是被哪個線程上鎖成功了，我服務應該保證這個線程執行的安全性，怎麼辦？鎖續命。

什麼意思，一旦這把鎖出現了上鎖操作，就意味着這把鎖開始投入使用，這時我的服務中需要有一個 daemon 線程定時去守護我的鎖的安全性，怎麼守護？

比如說鎖超時時間設置的是 1s，那麼我這個定時任務是每隔 300ms 去 redis 服務端做一次檢查，如果我還持有，你就給我續命，就像 session 會話的活躍機制一樣。

看個例子，我上鎖時候超時時間設置的是 1s，實際方法執行時間是 3s，這中間我的定時線程每隔 300ms 就會去把這把鎖的超時時間重新設置爲 1s，每隔 300ms 一次，成功將鎖續命成功。

public class RedisLockIdleThreadPool {
    private String threadAddLife_lua = "if redis.call(\"exists\", KEYS[1]) == 1\n" + 
        "then\n" + 
        "    return redis.call(\"expire\", KEYS[1], ARGV[1])\n" + 
        "else\n" + 
        "    return 0\n" + 
        "end";
  private volatile ScheduledExecutorService scheduledThreadPool;
  public RedisLockIdleThreadPool() {
    if (scheduledThreadPool == null) {
      synchronized (this) {
        if (scheduledThreadPool == null) {
          scheduledThreadPool = Executors.newSingleThreadScheduledExecutor(); // 我這樣創建線程池是爲了代碼的易讀性，大家務必使用ThreadPoolExecutor去創建
          scheduledThreadPool.scheduleAtFixedRate(() -> {
            addLife();
          }, 0, 300, TimeUnit.MILLISECONDS);
        }
      }
    }
  }
  private void addLife() {
            // ... 省略jedis的初始化過程
    List<byte[]> keys = Arrays.asList(RedisLock.lock_key.getBytes());
    List<byte[]> args = Arrays.asList(String.valueOf(1).getBytes());
    jedis.eval(threadAddLife_lua.getBytes(), keys, args);
  }
}

這就行嗎？還不行！

爲啥？想一下，如果每個服務中都像這樣去續命鎖，假如說 A 服務還在執行過程中的時候，還沒有執行完，就是說還沒有手動釋放鎖的時候，宕機，此時 redis 中鎖還在有效期。

服務 B 也一直在續命這把鎖，此時這把鎖一直在續命，但是 B 的這個續命一直續的是 A 當時設的鎖，這不是扯嗎？我自己在不斷續命，導致我的服務上一直獲取不到鎖，實際上 A 已經宕機了呀！該釋放了，不應該去續命了，這不是我服務 B 該乾的活！

續命的前提是，得判斷是不是當前進程持有的鎖，也就是我們的 APP_ID，如果不是就不進行續命。

續命鎖的 lua 腳本發生改變，如下：

  THREAD_ADD_LIFE("local v=redis.call(\"get\", KEYS[1]) \n" +   // get key
        "if v==false \n" +  // 如果不存在key，讀出結果v是false
        "then \n" +     // 不存在不處理
        "else \n" + 
        "    local match = string.find(v, ARGV[1]) \n" + // 存在，判斷是否能和APP_ID匹配，匹配不上時match是nil
        "    if match==\"nil\" \n" + 
        "    then \n" + 
        "    else  \n" + 
        "        return redis.call(\"expire\", KEYS[1], ARGV[2]) \n" + // 匹配上了返回的是索引位置，如果匹配上了意味着就是當前進程佔有的鎖，就延長時間
        "    end \n" + 
        "end")

3-13、別急，還有。鎖在我手裏，我掛了，這...

上邊 3-2 的問題在這裏說下，爲啥在上邊不說？是怕你剛開始就鑽到了這個問題中，導致本末倒置思維受限了。

即便我設置了一個很合理的 expire，比如 10s，但是線上如果真出現了 A 節點剛拿到鎖就宕機了，那其他節點也只能乾等着 10s 之後再去幹活了。

主要還是看你的業務能不能接受。而如果是 To C 的業務中，大部分場景無法接受的，因爲可能會導致用戶流失，獲得一個用戶多難呀。

所以我們需要另外一個監控服務，定時去監控 redis 中鎖的獲得者的健康狀態，如果獲取者超過 n 次無法通信，由監控服務負責將鎖摘除掉，讓別的線程繼續去獲取到鎖去幹活。

當然，這又引入了通信保證性的問題，如果監控服務和服務節點之間通信出現問題，那將導致很嚴重的後果。具體業務具體應用吧。

可以參考 zookeeper 分佈式鎖的設計，不在本文範圍內，暫不細聊。

風險！主從部署引來的問題

哨兵主從部署的時候，會存在一個風險問題，因爲 Redis 默認的主從複製是異步的，那很自然可以想到一個問題，極端情況下，如果剛往 master 節點寫入一個分佈式鎖，而這個指令流還沒有來得及同步給任意一個 slave 節點，此時，master 節點宕機，其中一個 slave 被哨兵選舉爲 master，此時是沒有這個鎖的，別的線程再次來獲取鎖，又獲取鎖成功了。

當然，這個概率極低，但是我們必須得承認這個風險的存在。其實，即使 redis 部署是單節點的話也會存在問題，如果 redis.conf 的相關持久化機制不合理，另外操作系統再配置一些影響參數，都會造成未持久化到磁盤中時，發生宕機且數據丟失，論風險的話如果爲了技術而質疑技術這將失去了意義。

這塊先不展開了，後面有時間我和大家聊聊 Redis 持久化、集羣的有意思的事。

從 Redis 官方文檔上摘抄如下（Replication – Redis）：

Redis uses by default asynchronous replication, which being low latency and high performance, is the natural replication mode for the vast majority of Redis use cases.

譯文：Redis 默認使用異步複製，低延遲和高性能，絕大多數的 Redis 服務使用自然複製模式。

完工，我建議你合上屏幕，自己在腦子裏重新過一遍，每一步都在做什麼，爲什麼要做，解決什麼問題。

想清楚之後，一定要，一定要自己親手來一遍代碼。本文用了很多的 lua 腳本，有同學如果看不懂 lua 腳本的不用着急，lua 腳本不用專門花時間去研究，只需要在用的時候，去找資料看看，現學現用即可。在這裏 lua 只是一種工具語言，隨時查隨時學就行。

【總結】本文我們一起從頭到尾梳理了一遍 Redis 分佈式鎖中的各種門道，其實很多點是不管用什麼做分佈式鎖都會存在的問題，重要的是思考的過程。

博客裏邊的代碼只是 demo，不接受吐槽哈哈哈。有時間我們再聊聊 zk 作鎖的事情。但是，本質上的幾個問題點本文已經拋出來了。

當然，隨着我們的積累越來越多，可能還會發現其中存在的一些問題，重要的是嘗試着這樣去思考問題，提升的就不只是技術點，而是技術面。

另外，千萬別爲了技術而技術，不要想入非非，要儘可能的貼近業務場景去想，比如你的產品是一把鐵鍬，但你一直在想如果有人用它來切菜就不好用了...... 將毫無意義。

也不要覺得一下就可以完美，要知道有些時候完成比完美更重要，何況，不存在完美。努力改變自己和身邊人的生活。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/cr29ivkRzOwlMUtgramcSw

猜你喜歡