掌握Redis分佈式鎖的正確姿式

本文中案例都會在上傳到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：

•  

public interface GrabService {
    /**     * 商品搶單     * @param orderId     * @param driverId     * @return     */    public ResponseResult grabOrder(int orderId, int driverId);}

GrabJvmLockServiceImpl：

•  

@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 :

•  

public interface OrderService {  public boolean grab(int orderId, int driverId);}

OrderServiceImpl :

•  

@Servicepublic 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優化

項目啓動順序：先啓動 Server-eureka註冊中心、在啓動 8004和8005端口

測試結果：這裏咱們能夠看到 8004 服務和 8005 服務 同時都有一個用戶去下單成功這個商品，可是這個商品只能有一個用戶可以去搶到，所以jvm鎖若是是在集羣或分佈式下，是沒法保證訪問共享變量的數據同時只有一個線程訪問的，沒法解決分佈式，集羣環境的問題。因此須要使用到分佈鎖。

分佈式鎖三種實現方式

分佈式鎖的實現方式總共有三種：

• 基於數據庫實現分佈式鎖

• 基於緩存（Redis）實現分佈式鎖

• 基於Zookeeper實現分佈式鎖

今天，咱們主要講的是基於Redis實現的分佈式鎖

reids實現分佈式鎖有三種方式

一、基於redis的 SETNX 實現分佈式鎖 二、Redisson實現分佈式鎖 四、使用redLock實現分佈式鎖

目錄結構：

方式一：基於 SETNX 實現分佈式鎖

將key的值設爲value ，當且僅當key不存在。若給定的key已經存在，則SETNX不作任何動做。setnx：當key存在，不作任何操做，key不存在，才設置

加鎖：

SET orderId driverId NX PX 30000 上面的命令若是執行成功，則客戶端成功獲取到了鎖，接下來就能夠訪問共享資源了；而若是上面的命令執行失敗，則說明獲取鎖失敗。

釋放鎖：關鍵，判斷是否是本身加的鎖。

GrabService ：

•  

public interface GrabService {
    /**     * 商品搶單     * @param orderId     * @param driverId     * @return     */    public ResponseResult grabOrder(int orderId, int driverId);}

GrabRedisLockServiceImpl ：

•  

@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實現分佈式鎖

流程圖：代碼實現：

•  

@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

目的：對共享資源作互斥訪問

所以antirez提出了新的分佈式鎖的算法Redlock，它基於N個徹底獨立的Redis節點（一般狀況下N能夠設置成5），意思就是N個Redis數據不互通，相似於幾個陌生人

代碼實現：

•  

@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 ，也包含過時時間(好比 PX30000 ，即鎖的有效時間)。爲了保證在某個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實現分佈式鎖，喜歡的小夥伴能夠關注我，對本文內容有疑問或者問題的同窗能夠留言，小農看到了會第一時間回覆，謝謝你們，你們加油