背景

如今的服務大多數是集群化部署，這就使得像synchronized、ReentrantLock等傳統的本地鎖失去了功效。因此需要引入第三方的服務實現對這些并發進程的調度，從而控制對共享資源的訪問，像redis、zookeeper、mysql等。其中又以redis的應用最為廣泛。

分布式鎖的要素

最重要的兩個要素：排他性、容錯性。

排他性是指在分布式應用集群中，同一個方法在同一時間內只能被一臺機器上的一個線程執行。

容錯性是指不論正常的業務執行完成，還是突發性的程序崩潰或者網絡中斷，都要保證分布式鎖最終一定能得到釋放，不能出現死鎖現象。

redis分布式鎖的基本命令

1、加鎖 SETNX key value

setnx 的含義就是 SET if Not Exists，有兩個參數 setnx(key, value)，該方法是原子性操作。如果 key 不存在，則設置當前 key 成功，返回 1；如果當前 key 已經存在，則設置當前 key 失敗，返回 0。

2、解鎖 del (key)

得到鎖的線程執行完任務，需要釋放鎖，以便其他線程可以進入。

3、配置鎖超時 expire (key，30s）

客戶端崩潰或者網絡中斷，資源將會永遠被鎖住,即死鎖，因此需要給key配置過期時間，以保證即使沒有被顯式釋放，這把鎖也要在一定時間后自動釋放。

OK，有了上面的理論基礎，我們就可以來逐步的揭開redis分布式鎖的神秘面紗。

我們以常見的扣減庫存的場景為例，當有線程來執行扣減庫存的方法時，大致邏輯是先判斷當前庫存，如果還有庫存的話，就庫存減1，然后生成明細記錄。

一把問題很多的鎖

首先看一段偽代碼。

methodA(){
  //ID為666的商品庫存扣減key
  String key = "stock:deduct:666"

  if（setnx（key，1） == 1）{
      expire(key,10,TimeUnit.SECONDS)
      try {
          //查詢是否有庫存
          //扣減庫存
          //生成明細記錄
      } finally {
          del（key）
      }
  }else{
    //獲取鎖失敗，睡眠100毫秒，然后自旋調用本方法
    methodA()
  }
}

這段代碼的主要邏輯是，先給ID為666的商品庫存上鎖，然后設置key的過期時間為10秒，之后就執行扣減庫存的邏輯了，等業務邏輯執行完成，就刪除key釋放鎖。在此期間如果有其他線程來獲取鎖，會上鎖失敗，失敗后就等一會再次調用methodA方法繼續嘗試上鎖，然后循環往復，直到上鎖成功。

看上去大功告成了，所謂的分布式鎖也不過如此。

然而，正如我們標題上寫的，這是一把問題很多的鎖，有什么問題呢？

首先最大的問題是，多個命令之間不是原子操作。在setnx和expire之間是分了兩步來執行的，如果setnx成功，但是expire卻執行失敗，或者還沒有執行就突發宕機，就造成了這個資源的死鎖，違反了我們上面提到的容錯性原則。

另外存在的一個問題是，可能會出現線程A刪掉了線程B的鎖。假設有兩個線程A和B，A先上鎖成功開始執行業務邏輯，但由于某些原因導致A執行很慢，15秒才執行完，但A的鎖有效期只有10秒，A鎖過期后，B上鎖成功，但是B還沒有執行完業務邏輯，線程A業務邏輯執行完成，執行刪鎖操作，此時刪除的，實際上是B的鎖，B的鎖刪掉了，也就無法阻止其他線程來加鎖，違反了上面提到的排他性原則。

如何解決這兩個問題呢？

優化后的鎖

第一個問題，既然多個命令之間不是原子操作，我們用一個命令就行了，而redis恰好也提供了一個這樣的命令，setex，即在賦值的時候設置過期時間，這是一個原子命令。對應到java中，也有這樣的API供我們使用：

redisTemplate.opsForValue().setIfAbsent("key","success",10,TimeUnit.SECONDS)

第二個問題，可以在刪除鎖之前做一個判斷，驗證當前要刪除的鎖是不是自己的鎖，實現方式也很簡單，可以將value值設置為當前的線程ID或者隨便一個UUID。

優化后的偽代碼應該是這樣的：

methodA(){
  //ID為666的商品庫存扣減key
  String key = "stock:deduct:666";
  String value = Thread.currentThread().getId();

  if（setex（key, 10, value） == 1）{
      try {
          //查詢是否有庫存
          //扣減庫存
          //生成明細記錄
      } finally {
        if(get(key).equals(value)){
          del（key）
        }
      }
  }else{
    //獲取鎖失敗，睡眠100毫秒，然后自旋調用本方法
    methodA()
  }
}

這把鎖總沒問題了吧？

然而，細細考究一下，還是會發現不妥之處。雖然我們刪除鎖的時候做了判斷，但仍有可能刪錯鎖。根本原因是判斷鎖和刪除鎖同樣不是原子操作。

那到底如何保證絕對的原子性？

lua腳本的橫空出世

這里我們不去深究lua腳本是什么，只需要知道，lua是一個腳本語言，redis執行lua腳本的時候，會將它里面的命令當做一個整體去執行，要么全部執行成功，要么出現異常，結果不會更新到redis中。

因此，上面的刪鎖操作，我們完全可以將判斷命令和刪除命令都放到lua腳本中，然后由代碼去執行lua腳本，最終會實現我們想要的原子操作。

實際上，這也正是redis官方推薦的做法。具體可查看官方文檔：??set 命令 -- Redis中國用戶組（CRUG）??。

這里提供一段java中調用lua腳本的代碼，大家看了后可以加深理解：

String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
Integer result = redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class), Arrays.asList(lockKey), uuid);

其中的腳本代碼是官方文檔中提供的，可以直接復制過來使用。

原生分布式鎖

綜合上面所說的，一個完整的原生分布式鎖應該就是下面這個樣子了：

methodA(){
  //ID為666的商品庫存扣減key
  String key = "stock:deduct:666";
  String value = Thread.currentThread().getId();
  String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
  if（setex（key, 10, value） == 1）{
      try {
          //查詢是否有庫存
          //扣減庫存
          //生成明細記錄
      } finally {
        //解鎖
        Long result = redisTemplate.execute( new DefaultRedisScript<>(script,Long.class),Arrays.asList(lockKey),uuid);
      }
  }else{
    //獲取鎖失敗，睡眠100毫秒，然后自旋調用本方法
    methodA()
  }
}

這就是一個比較完善的分布式鎖了，既滿足了對共享資源的并發控制，又保證了加鎖、解鎖的原子性操作，防止突發狀況造成的死鎖問題。

這里大家再想一個問題，如何避免業務執行時間過長鎖過期的問題？為了保證排他性，肯定要保證在業務執行時間內，鎖是一定不能過期的。在原生的分布式鎖中，沒有什么好的方法，只能加長鎖的過期時間，保證業務一定能執行完成。

那么，有沒有更好的解決方案呢？

Hi，我叫redisson

redisson是redis官方推薦的一個分布式鎖的框架，它幫我們解決了上面提到的所有問題，底層也是用了lua腳本實現，同時又提供了watchdog(看門狗)機制，在鎖將要過期的時候，會自動檢測業務是否執行完成，如果沒有完成，則自動延長鎖的過期時間，直到業務執行完成。而且最重要的一點，使用起來非常簡單，幾行代碼就可以搞定，不像原生鎖那樣繁瑣，是我們進行分布式鎖開發的不二選擇。這里不做詳細描述了，感興趣的可以在網上搜索一下。

好了，關于redis分布式鎖就到這里了。