> 本文中案例都會在上傳到git上，請放心瀏覽 > git地址：https://github.com/muxiaonong/Spring-Cloud/tree/master/order-lock > 本文會使用到 三臺 redis 獨立伺服器，可以自行提前搭建好 ## 前言 在Java中，我們對於鎖會比較熟悉，常用的有 ```synchronized、Lock```鎖，在java併發程式設計中，我們通過鎖，來實現當多個執行緒競爭同一個共享資源或者變數而造成的資料不一致的問題，但是JVM鎖只能針對於單個應用服務，隨著我們業務的發展需要，單體單機部署的系統早已演化成分散式系統，由於分散式系統的多執行緒、多程序而且分佈在不同的機器上，這個時候JVM鎖的併發控制就沒有效果了，為了解決跨JVM鎖並且能夠控制共享資源的訪問，於是有了分散式鎖的誕生。  ## 什麼是分散式鎖 分散式鎖是控制分散式系統之間同步訪問共享資源的一種方式。在分散式系統中，常常需要協調他們的動作。如果不同的系統或是同一個系統的不同主機之間共享了一個或一組資源，那麼訪問這些資源的時候，往往需要互斥來防止彼此干擾來保證一致性，在這種情況下，便需要使用到分散式鎖 ## 為什麼JVM鎖在分散式下不可以呢？ 我們通過程式碼來看一下就知道，為什麼叢集下jvm鎖是不可靠的呢？我們模擬一下商品搶購的場景，A服務有十個使用者去搶購這個商品，B服務有十個使用者去搶購這個商品，當有其中一個使用者搶購成功後，其他使用者不可以在對這個商品進行下單操作，那麼到底是A服務會搶到還是B服務會搶到這個商品呢，我們來看一下 **當其中有一個使用者搶購成功後，status會變成1**  **GrabService：** ```java public interface GrabService { /** * 商品搶單 * @param orderId * @param driverId * @return */ public ResponseResult grabOrder(int orderId, int driverId); } ``` **GrabJvmLockServiceImpl：** ```java @Service("grabJvmLockService") public class GrabJvmLockServiceImpl implements GrabService { @Autowired OrderService orderService; @Override public ResponseResult grabOrder(int orderId, int driverId) { String lock = (orderId+""); synchronized (lock.intern()) { try { System.out.println("使用者:"+driverId+" 執行下單邏輯"); boolean b = orderService.grab(orderId, driverId); if(b) { System.out.println("使用者:"+driverId+" 下單成功"); }else { System.out.println("使用者:"+driverId+" 下單失敗"); } } finally { } } return null; } } ``` **OrderService :** ```java public interface OrderService { public boolean grab(int orderId, int driverId); } ``` **OrderServiceImpl :** ```java @Service public class OrderServiceImpl implements OrderService { @Autowired private OrderMapper mapper; public boolean grab(int orderId, int driverId) { Order order = mapper.selectByPrimaryKey(orderId); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } if(order.getStatus().intValue() == 0) { order.setStatus(1); mapper.updateByPrimaryKeySelective(order); return true; } return false; } } ``` 這裡我們模擬叢集環境，啟動兩個埠，8004和8005進行訪問 這裡我們用**jmeter**進行測試 如果不會**jmeter**的可以看我之前對tomcat進行壓測的文章：[tomcat優化](https://blog.csdn.net/qq_14996421/article/details/105887239) 專案啟動順序：先啟動 Server-eureka註冊中心、在啟動 8004和8005埠   **測試結果：**  這裡我們可以看到 8004 服務和 8005 服務 **同時都有一個使用者去下單成功這個商品**，但是這個商品只能有一個使用者能夠去搶到，因此jvm鎖如果是在叢集或分散式下，是無法保證訪問共享變數的資料同時只有一個執行緒訪問的，無法解決分散式，叢集環境的問題。所以需要使用到分佈鎖。 ## 分散式鎖三種實現方式 **分散式鎖的實現方式總共有三種：** - 基於資料庫實現分散式鎖 - 基於快取（Redis）實現分散式鎖 - 基於Zookeeper實現分散式鎖 今天，我們主要講的是基於Redis實現的分散式鎖 ## reids實現分散式鎖有三種方式 1、基於redis的 SETNX 實現分散式鎖 2、Redisson實現分散式鎖 4、使用redLock實現分散式鎖 **目錄結構：**  #### 方式一：基於 SETNX 實現分散式鎖 將key的值設為value ，當且僅當key不存在。 若給定的key已經存在，則SETNX不做任何動作。 **setnx：當key存在，不做任何操作，key不存在，才設定** **加鎖：** >SET orderId driverId NX PX 30000 >上面的命令如果執行成功，則客戶端成功獲取到了鎖，接下來就可以訪問共享資源了；而如果上面的命令執行失敗，則說明獲取鎖失敗。 **釋放鎖：** 關鍵，判斷是不是自己加的鎖。 **GrabService ：** ```java public interface GrabService { /** * 商品搶單 * @param orderId * @param driverId * @return */ public ResponseResult grabOrder(int orderId, int driverId); } ``` **GrabRedisLockServiceImpl ：** ```java @Service("grabRedisLockService") public class GrabRedisLockServiceImpl implements GrabService { @Autowired StringRedisTemplate stringRedisTemplate; @Autowired OrderService orderService; @Override public ResponseResult grabOrder(int orderId , int driverId){ //生成key String lock = "order_"+(orderId+""); /* * 情況一，如果鎖沒執行到釋放，比如業務邏輯執行一半，運維重啟服務，或 伺服器掛了，沒走 finally，怎麼辦？ * 加超時時間 */ // boolean lockStatus = stringRedisTemplate.opsForValue().setIfAbsent(lock.intern(), driverId+""); // if(!lockStatus) { // return null; // } /* * 情況二：加超時時間,會有加不上的情況，運維重啟 */ // boolean lockStatus = stringRedisTemplate.opsForValue().setIfAbsent(lock.intern(), driverId+""); // stringRedisTemplate.expire(lock.intern(), 30L, TimeUnit.SECONDS); // if(!lockStatus) { // return null; // } /* * 情況三：超時時間應該一次加，不應該分2行程式碼， * */ boolean lockStatus = stringRedisTemplate.opsForValue().setIfAbsent(lock.intern(), driverId+"", 30L, TimeUnit.SECONDS); if(!lockStatus) { return null; } try { System.out.println("使用者:"+driverId+" 執行搶單邏輯"); boolean b = orderService.grab(orderId, driverId); if(b) { System.out.println("使用者:"+driverId+" 搶單成功"); }else { System.out.println("使用者:"+driverId+" 搶單失敗"); } } finally { /** * 這種釋放鎖有，可能釋放了別人的鎖。 */ // stringRedisTemplate.delete(lock.intern()); /** * 下面程式碼避免釋放別人的鎖 */ if((driverId+"").equals(stringRedisTemplate.opsForValue().get(lock.intern()))) { stringRedisTemplate.delete(lock.intern()); } } return null; } } ``` 這裡可能會有人問，如果我業務的執行時間超過了鎖釋放的時間，會怎麼辦呢？我們可以使用守護執行緒，只要我們當前執行緒還持有這個鎖，到了10S的時候，守護執行緒會自動對該執行緒進行加時操作，會續上30S的過期時間，直到把鎖釋放，就不會在進行續約了，開啟一個子執行緒，原來時間是N，每隔N/3，在去續上N **關注點：** 1. key，是我們的要鎖的目標，比如訂單ID。 2. driverId 是由我們的商品ID，它要保證在足夠長的一段時間內在所有客戶端的所有獲取鎖的請求中都是唯一的。即一個訂單被一個使用者搶。 3. NX表示只有當orderId不存在的時候才能SET成功。這保證了只有第一個請求的客戶端才能獲得鎖，而其它客戶端在鎖被釋放之前都無法獲得鎖。 4. PX 30000表示這個鎖有一個30秒的自動過期時間。當然，這裡30秒只是一個例子，客戶端可以選擇合適的過期時間。 5. **這個鎖必須要設定一個過期時間。** 否則的話，當一個客戶端獲取鎖成功之後，假如它崩潰了，或者由於發生了網路分割槽，導致它再也無法和Redis節點通訊了，那麼它就會一直持有這個鎖，而其它客戶端永遠無法獲得鎖了。antirez在後面的分析中也特別強調了這一點，而且把這個過期時間稱為鎖的有效時間(lock validity time)。獲得鎖的客戶端必須在這個時間之內完成對共享資源的訪問。 6. 此操作不能分割。 >SETNX orderId driverId EXPIRE orderId 30 雖然這兩個命令和前面演算法描述中的一個SET命令執行效果相同，但卻不是原子的。如果客戶端在執行完SETNX後崩潰了，那麼就沒有機會執行EXPIRE了，導致它一直持有這個鎖。造成死鎖。 #### 方式二：基於redisson實現分散式鎖 **流程圖：**  程式碼實現： ```java @Service("grabRedisRedissonService") public class GrabRedisRedissonServiceImpl implements GrabService { @Autowired RedissonClient redissonClient; @Autowired OrderService orderService; @Override public ResponseResult grabOrder(int orderId , int driverId){ //生成key String lock = "order_"+(orderId+""); RLock rlock = redissonClient.getLock(lock.intern()); try { // 此程式碼預設 設定key 超時時間30秒，過10秒，再延時 rlock.lock(); System.out.println("使用者:"+driverId+" 執行搶單邏輯"); boolean b = orderService.grab(orderId, driverId); if(b) { System.out.println("使用者:"+driverId+" 搶單成功"); }else { System.out.println("使用者:"+driverId+" 搶單失敗"); } } finally { rlock.unlock(); } return null; } } ``` 關注點： 1. redis故障問題。 如果redis故障了，所有客戶端無法獲取鎖，服務變得不可用。為了提高可用性。我們給redis 配置主從。當master不可用時，系統切換到slave，由於Redis的主從複製（replication）是非同步的，這可能導致喪失鎖的安全性 >1.客戶端1從Master獲取了鎖。 2.Master宕機了，儲存鎖的key還沒有來得及同步到Slave上。 3.Slave升級為Master。 4.客戶端2從新的Master獲取到了對應同一個資源的鎖。 客戶端1和客戶端2同時持有了同一個資源的鎖。鎖的安全性被打破。 2. 鎖的有效時間(lock validity time)，設定成多少合適？如果設定太短的話，鎖就有可能在客戶端完成對於共享資源的訪問之前過期，從而失去保護；如果設定太長的話，一旦某個持有鎖的客戶端釋放鎖失敗，那麼就會導致所有其它客戶端都無法獲取鎖，從而長時間內無法正常工作。應該設定稍微短一些，如果執行緒持有鎖，開啟執行緒自動延長有效期 #### 方式三：基於RedLock實現分散式鎖 針對於以上兩點，antirez設計了Redlock演算法 Redis的作者antirez給出了一個更好的實現，稱為Redlock，算是Redis官方對於實現分散式鎖的指導規範。Redlock的演算法描述就放在Redis的官網上： [https://redis.io/topics/distlock](https://redis.io/topics/distlock) **目的：對共享資源做互斥訪問** 因此antirez提出了新的分散式鎖的演算法Redlock，**它基於N個完全獨立的Redis節點**（通常情況下N可以設定成5），意思就是N個Redis資料不互通，類似於幾個陌生人 **程式碼實現：** ```java @Service("grabRedisRedissonRedLockLockService") public class GrabRedisRedissonRedLockLockServiceImpl implements GrabService { @Autowired private RedissonClient redissonRed1; @Autowired private RedissonClient redissonRed2; @Autowired private RedissonClient redissonRed3; @Autowired OrderService orderService; @Override public ResponseResult grabOrder(int orderId , int driverId){ //生成key String lockKey = (RedisKeyConstant.GRAB_LOCK_ORDER_KEY_PRE + orderId).intern(); //紅鎖 RLock rLock1 = redissonRed1.getLock(lockKey); RLock rLock2 = redissonRed2.getLock(lockKey); RLock rLock3 = redissonRed2.getLock(lockKey); RedissonRedLock rLock = new RedissonRedLock(rLock1,rLock2,rLock3); try { rLock.lock(); // 此程式碼預設 設定key 超時時間30秒，過10秒，再延時 System.out.println("使用者:"+driverId+" 執行搶單邏輯"); boolean b = orderService.grab(orderId, driverId); if(b) { System.out.println("使用者:"+driverId+" 搶單成功"); }else { System.out.println("使用者:"+driverId+" 搶單失敗"); } } finally { rLock.unlock(); } return null; } } ```  執行Redlock演算法的客戶端依次執行下面各個步驟，來完成 獲取鎖 的操作： 1. 獲取當前時間（毫秒數）。 2. 按順序依次向N個Redis節點執行 **獲取鎖** 的操作。這個獲取操作跟前面基於單Redis節點的 **獲取鎖** 的過程相同，包含value driverId ，也包含過期時間(比如 `PX 30000` ，即鎖的有效時間)。為了保證在某個Redis節點不可用的時候演算法能夠繼續執行，這個 **獲取鎖** 的操作還有一個超時時間(time out)，它要遠小於鎖的有效時間（幾十毫秒量級）。 3. 客戶端在向某個Redis節點獲取鎖失敗以後，應該立即嘗試下一個Redis節點。這裡的失敗，應該包含任何型別的失敗，比如該Redis節點不可用，或者該Redis節點上的鎖已經被其它客戶端持有 4. 計算整個獲取鎖的過程總共消耗了多長時間，計算方法是用當前時間減去第1步記錄的時間。如果客戶端從大多數Redis節點（**>= N/2+1**）成功獲取到了鎖，比如：五臺機器如果加鎖成功三臺就預設加鎖成功，並且獲取鎖總共消耗的時間沒有超過鎖的有效時間(lock validity time)，那麼這時客戶端才認為最終獲取鎖成功；否則，認為最終獲取鎖失敗 5. 如果最終獲取鎖成功了，那麼這個鎖的有效時間應該重新計算，它等於最初的鎖的有效時間減去第3步計算出來的獲取鎖消耗的時間。 6. 如果最終獲取鎖失敗了（可能由於獲取到鎖的Redis節點個數少於N/2+1，或者整個獲取鎖的過程消耗的時間超過了鎖的最初有效時間），那麼客戶端應該立即向所有Redis節點發起 **釋放鎖** 的操作（即前面介紹的Redis Lua指令碼）。 上面描述的只是 獲取鎖 的過程，而 釋放鎖 的過程比較簡單：客戶端向所有Redis節點發起 釋放鎖 的操作，不管這些節點當時在獲取鎖的時候成功與否。 ## 總結 到這裡redis分散式鎖就講完了，具體使用哪一種型別的分散式鎖需要看公司業務的，流量大的可以使用RedLock實現分散式鎖，流量小的可以使用redisson，後面會講解Zookeeper實現分散式鎖，喜歡的小夥伴可以關注我，對本文內容有疑問或者問題的同學可以留言，小農看到了會第一時間回覆，謝謝大家，大