# 前言 本文力爭以最簡單的語言，以博主自己對分散式鎖的理解，按照自己的語言來描述分散式鎖的概念、作用、原理、實現。如有錯誤，還請各位大佬海涵，懇請指正。分散式鎖分兩篇來講解，本篇講解客戶端，下一篇講解redis服務端。 # 概念 如果把分散式鎖的概念搬到這裡，博主也會覺得枯燥。博主這裡以舉例的形式來描繪它。 試想一種場景，在一個偏遠小鎮上的火車站，只有一個售票視窗。 火車站來了10名旅客，前往售票視窗購買火車票，旅客只能排隊購票，排到第一的旅客，可以與售票員溝通，買票。 好啦，以上就是一個分散式鎖的場景，我們來分析一下每一個細節。 每位旅客可以理解為一個系統或者執行緒。他們在競爭售票員的工作時間。  是不是覺得分散式鎖也不是什麼高大上的概念。有同學會問，鎖到底在哪裡呢？還是買票場景，我們看看鎖長什麼樣子。  我們深入想一下，這10位旅客本來是並行的（沒有買票前，他們有的在吃飯，有的在玩手機，等等等），而到了買票的時候，就必須排隊（序列），而不是一起買票。 沒錯，就是在特定的場景下，將並行的場景，變成序列，就是分散式鎖的奧義所在。 # 作用 分散式鎖的作用不但非常大，而且非常多。 在軟體設計中，比如電商秒殺活動。商家預備了1000件貨物，也就只有這1000件貨，有1500人蔘與秒殺，可以理解為1500個執行緒來排隊購買商品。那就必須將這1500個執行緒排個隊（比如按照時間），設定一把鎖，一個購買過程結束，再開始下一個。 為什麼redis可以實現分散式鎖呢？ 我們以購票舉例，購票視窗前的這個鎖，是每位旅客都可以看到的。 這裡我們可以得出一個結論，一把鎖首先要具有的屬性是：想要獲得鎖的人都可以看到。 這把鎖既不能屬於伺服器A，也不能屬於伺服器B，因為他們都不知道另一方的存在，那就必須選擇一個公信的第三方來作為鎖。當~當~當~ redis閃亮登場。當然zookeeper也可以實現，這裡先挖一個坑，以後再填zookeeper吧。  # 原理 ## 加鎖的基本思路 redis中有一條指令非常有意思，它叫做setnx  當redis中不存在key值為“lock”的時候，可以設定成功；當存在key值時，設定失敗。 這句指令，好比是，詢問一下，到我買票了嗎？返回結果是1的時候，到您買票了；返回結果是0的時候，還沒到您，稍後再詢問。 我們的鎖過程可以這樣來操作： - setnx lock 鎖值 - 處理業務邏輯 - 釋放鎖 del lock ## 優化一 為什麼要優化？ 試想，如果setnx lock 1 加鎖成功，這個時候系統因為其他原因，掛掉了，就永遠無法執行del lock了。 要避免這種情況，怎麼辦呢？給鎖一個過期時間。  這樣無論系統是否宕機，都會在10秒後釋放鎖。看似很美好，雖然setnx lock 1 與 expire lock 10之間的時間間隙非常小，但仍然有風險，加入系統執行完 setnx lock 1 後，宕機了，並沒有執行 過期指令 expire lock 10，再次產生了一把無法解開的鎖，“死鎖”。 這時候引入了一個概念，叫做原子操作。即這兩條指令需要在一個原子操作內執行完成。 ``` set key value [expiration EX seconds|PX milliseconds] [NX|XX] ```  ## 優化二 why?上一個優化已經把上鎖過程做成了原子操作，還需要什麼優化呢？ 當然有，試想一下，之前程式碼set lock 1 ex 10 nx，設定過期時間是10秒，那麼這個10秒是否可靠呢？顯然不可靠。 我們加鎖的過程是 加鎖---執行業務程式碼---釋放鎖 加入業務程式碼的執行時間超過10秒呢？是不是業務程式碼還沒有執行完，鎖就已經釋放了。放在購票場景中，第一位旅客還沒有完成購票，第二位旅客就開始購票。顯然不合理。怎麼辦呢？ 這裡我們需要估計業務程式碼的執行時間，加入預估出來的時間是10秒，可以在業務程式碼中開闢一個“續命”的操作。 - 加鎖 set lock 1 ex 10 nx - 每過3秒，把該鎖的時間重新設定為 10秒 - 執行業務程式碼 - 釋放鎖 del lock 這裡的續命時間間隔 = 過期時間 10S / 3 這樣設定比較合理，可以防止一次續命失敗。 ## 優化三 納尼?還有問題嗎？ 有，而且可以算是一個bug，我們一直在用 set lock 1 ex 10 nx 來加鎖，用del lock 來釋放鎖。 我們需要明確知道，釋放的鎖，是自己加上的。 可以set lock uuid ex 10 nx 來解決該問題。 ## 拓展-可重入鎖 一個執行緒獲取到鎖以後，再次獲取鎖，就是可重入鎖。 但博主現在遇到的問題，一般不需要可重入鎖即可解決。java中ReentrantLock就是可重入鎖。 可重入鎖，對程式碼的複雜度增加了很多，玩不好，容易扯襠。謹慎使用。 # 實現 已經講了很多優化相關的內容，這裡博主就直接寫優化後的程式碼了。 博主使用java來實現。而redis官方（https://redis.io/clients#java）推薦的有三個框架。分別是Jedis、lettuce、Redisson。 由於博主在本篇中主要討論單個redis的情況，而redisson主要用來處理分散式redis，下一篇博文使用redisson，敬請期待。  springboot2.x 預設採用了 lettuce，所以博主就使用lettuce來實現分散式鎖。 ## 引入依賴 ```xml org.springframework.boot spring-boot-starter-data-redis org.apache.commons commons-pool2 com.alibaba fastjson 1.2.72 ``` ## 配置檔案 既然要測試分散式鎖，那麼就至少應該跑兩份程式碼，所以配置檔案也應該是兩份，這裡博主偷個懶，提供一份配置檔案，另一份配置檔案修改下server的埠即可。 ```yaml server: port: 80 spring: redis: # redis的ip地址 host: redis的ip地址 # redis的埠號 port: 6379 # redis的密碼 password: 你的密碼 lettuce: pool: # 最大連結數 max-active: 30 # 連結池中最大空閒連結數 max-idle: 15 # 最大阻塞等待連結時長 預設不限制 -1 max-wait: 2000 # 最小空閒連結數 min-idle: 10 # 連結超時時長 shutdown-timeout: 10000 ``` ## lettuce配置類 這個類博主就不細講了，springboot整合lettuce，序列化博主更偏愛FastJson ```java import com.alibaba.fastjson.support.spring.GenericFastJsonRedisSerializer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; /** * @author xujp * redis 配置類 將RedisTemplate交給spring託管 */ @Configuration public class RedisConfig { @Bean public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory){ RedisTemplate redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(redisConnectionFactory); StringRedisSerializer stringRedisSerializer = new StringRedisSerializer(); GenericFastJsonRedisSerializer genericFastJsonRedisSerializer = new GenericFastJsonRedisSerializer(); redisTemplate.setKeySerializer(stringRedisSerializer); redisTemplate.setValueSerializer(genericFastJsonRedisSerializer); redisTemplate.setHashKeySerializer(stringRedisSerializer); redisTemplate.setHashValueSerializer(genericFastJsonRedisSerializer); redisTemplate.afterPropertiesSet(); return redisTemplate; } } ``` ## 分散式鎖 重頭戲來了，手寫分散式鎖的核心程式碼示例。 ```java import com.redis.demo1.thread.WatchDog; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; import java.util.UUID; import java.util.concurrent.TimeUnit; /** * @author xujp */ @RestController @RequestMapping("/test") public class TestController { @Autowired private RedisTemplate redisTemplate; @GetMapping public void lock(){ String uuid = UUID.randomUUID().toString(); //System.out.println(uuid); WatchDog watchDog; try { // 自旋 while (true) { // 嘗試獲取鎖 Boolean hasLock = redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3l, TimeUnit.SECONDS); if(hasLock) { // 看門狗“續命“ watchDog = new WatchDog(redisTemplate, uuid); watchDog.start(); // 業務邏輯start int num = (int) redisTemplate.opsForValue().get("num"); //Thread.sleep(4000); // 假設業務需要4s處理時間 redisTemplate.opsForValue().set("num", num - 1); System.out.println(num); // 業務邏輯處理 end break; }else{ // 睡眠100ms再自旋 Thread.sleep(100); } } }catch (Exception e){ System.out.println(e); }finally { // 關閉鎖 String l = (String) redisTemplate.opsForValue().get("lock"); if (l.equalsIgnoreCase(uuid)) { redisTemplate.delete("lock"); } } } } ``` 分散式鎖“續命”程式碼示例 ```java import org.springframework.data.redis.core.RedisTemplate; import java.util.concurrent.TimeUnit; /** * @author xujp */ public class WatchDog extends Thread { private RedisTemplate redisTemplate; private String uuid; public WatchDog(RedisTemplate redisTemplate, String uuid){ this.redisTemplate = redisTemplate; this.uuid = uuid; } public void run(){ // 續命邏輯 while (true){ try { // 獲取鎖的value Object redisUUID = redisTemplate.opsForValue().get("lock"); // 判斷當前父執行緒是否已經釋放鎖，如果父執行緒已釋放，則跳出執行緒 if(redisUUID==null || !redisUUID.toString().equals(uuid)){ break; } // 續命 redisTemplate.expire("lock", 3l, TimeUnit.SECONDS); // 沒隔1s續命一次 Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); break; } } } } ``` ## 測試 首先我們將程式碼分別以80和81埠run起來。 有精力的同學，還可以再搭建一個nginx將請求分流到80和81。這裡博主簡單粗暴地使用jmeter請求。 博主使用jmeter來測試，博主預設大家都會使用（不會使用的童鞋需要學習嘍）。 ### jmeter準備工作 在jmeter中設定50個執行緒  在該執行緒下設定兩個介面，分別請求80和81   ### redis準備工作 在redis中設定一對鍵值 num  至此，就可以在jmeter中開啟請求了 ### 測試結果 我們先來看redis中num的值  我們再分別檢視80和81的日誌   # 總結 本文講述了利用redis實現分散式鎖的原理，分散式鎖本質上是將併發請求按順序處理，那麼這把鎖就成為了所有請求的瓶頸，如何打破鎖的瓶頸呢？敬請關注博主，後續填坑（博主挖坑必填）。 本文留下的兩個坑： 1，為了使redis高可用，redis集群后，如何解決redis端因為網路問題導致鎖不同步問題？ 2，分散式鎖實現了併發排隊，鎖成為了效能瓶頸，如何提高效能？