概述

我們在實際的介面或者業務開發中，不管是伺服器單點還是伺服器叢集，都會有分散式鎖的使用場景。 比如最常見的介面重複提交（業務重複處理）、商品超賣等問題，通用的解決方案就是本文所使用的“ 分散式鎖”， 在同一個業務中，其中一個請求獲取到鎖之後，其他請求只有在獲取到鎖的請求釋放鎖（或者鎖失效）之後才能繼續“爭搶”鎖， 沒有獲得鎖的請求是沒有執行業務的許可權的。

方案論證

這裡我們主要討論兩種方案：基於redis的分散式鎖和基於zookeeper的分散式鎖

基於redis的分散式鎖

redis自身就提供了命令：SET key value NX PX expireTimeMs，專門用於處理分散式鎖的場景，效率高且提供鎖失效機制， 即使由於某種情況客戶端沒有傳送解鎖請求，也不會造成死鎖。

但是如果redis跑在叢集的情況下，由於redis叢集之間採用非同步的方式進行資料同步，因此在併發量大的情況下有可能遇到資料同步不及時造成多個請求同時獲取到鎖， 雖然業界有redlock演算法以及redisson客戶端實現能基本處理此類問題，也並不能完美解決這個問題，其演算法邏輯實現還很複雜， 更有甚者有分散式的專家Martin寫了一篇文章《How to do distributed locking》, 質疑redlock的正確性。Martin最後對redlock演算法的形容是： neither fish nor fowl （非驢非馬）。 本人覺得這篇文章（《基於Redis的分散式鎖真的安全嗎？》

），就redis叢集分散式鎖的安全問題就講得非常好。

結論：

• 優點：效能好
• 缺點：存在叢集資料同步不及時問題；鎖失效時間不好控制

因此，要想使用redis分散式鎖，最好使用redis單點模式，但是沒有人能保證redis單點的高可用性。

基於zookeeper的分散式鎖

zookeeper是一個分散式的，開放原始碼的分散式應用程式協調服務，是一個為分散式應用提供一致性服務的軟體， 提供的功能包括：配置維護、域名服務、分散式同步、組服務等。zookeeper機制規定同一個節點下只能有一個唯一名稱的節點， zookeeper上的一個znode看作是一把鎖，所有客戶端都去create同一個znode，最終成功建立的那個客戶端也即擁有了這把鎖。 zookeeper節點有兩大型別：持久化節點和臨時節點，客戶端建立一個臨時節點，當此客戶端與zookeeper server斷開後，該臨時節點會自動刪除。 由於zookeeper本身就強一致性的實現機制，因此不存在資料不一致的問題。

zookeeper提供了原生的API方式操作zookeeper，因為這個原生API使用起來並不是讓人很舒服，於是出現了zkclient這種方式，以至到後來出現了Curator框架， Curator對zkclient做了進一步的封裝，讓人使用zookeeper更加方便。有一句話，Guava is to JAVA what Curator is to Zookeeper。 Curator實現zookeeper分散式鎖的基本原理如下：

• 在zookeeper指定節點（${serviceLockName}）下建立臨時順序節點node_n
• 獲取${serviceLockName}下所有子節點children
• 對子節點按節點自增序號從小到大排序，判斷本節點是不是第一個子節點
• 若是，則獲取鎖；若不是，則監聽比該節點小的那個節點的刪除事件
• 若監聽事件生效，則回到第二步重新進行判斷，直到獲取到鎖
• 若超過等待時間，則獲取鎖失敗

就上面的Curator對分散式鎖實現的演算法還是挺複雜的，效率也不是太高，因為建立節點、獲取所有子節點並排序等操作涉及到多個網路IO以及程式碼邏輯處理，所以效率上會打折扣， 還有釋放鎖的時候只會刪除children節點，並不會刪除${serviceLockName}節點，因此zookeeper server中有可能會出現大量的${serviceLockName}節點佔用記憶體空間和Watcher。

因此，本人覺得Curator有些過於複雜了，可以直接利用zookeeper的特性（一個節點下只能有一個唯一名稱的節點，客戶端建立一個臨時節點，當此客戶端與zookeeper server斷開後，該臨時節點會自動刪除）， 重複建立子節點會丟擲KeeperException.NodeExistsException（節點已存在異常）來實現zookeeper分散式鎖。

結論：

• 優點：不存在資料不一致問題；有效的解決單點問題；鎖有效時間控制靈活
• 缺點：效能稍差，因為每次在建立鎖和釋放鎖的過程中，都要動態建立、銷燬臨時節點來實現鎖功能。並且建立和刪除節點只能通過Leader伺服器來執行，然後將資料同步到其他機器上。

因此，本文強烈推薦使用zookeeper來實現分散式鎖，但是又會多引入元件，為專案增加了風險。

專案原始碼

本專案程式碼已經託管到github與碼雲上：
github ：distributelock-spring-boot-starter
碼雲：distributelock-spring-boot-starter

使用方法

1. 分散式鎖starter jar包引用

<dependency>
   <groupId>cn.dslcode</groupId>
   <artifactId>distributelock-spring-boot-starter</artifactId>
   <version>1.0.0</version>
</dependency>

1. spring-data-redis或zookeeper的jar包引用，使用redis的話需要依賴RedisTemplate

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

或

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>${zookeeper.version}</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
        <exclusion>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
        </exclusion>
    </exclusions>
</dependency>

1. 配置引數

# 分散式鎖方式：redis或zookeeper
#distributelock.type=redis
distributelock.type=zookeeper
# 使用redis分散式鎖，配置redis連線
# spring.redis.host=127.0.0.1
# spring.redis.port=6379
# 使用zookeeper分散式鎖，配置zookeeper連線
distributelock.zookeeper.connect-string=127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183

1. 在需要進行分散式鎖控制的方法新增@Lockable註解，註解欄位如下

public @interface Lockable {

    /** lock key字首，每一個業務一個key */
    String key();

    /** 等待時間/毫秒 */
    int waitTimeMs() default 0;

    /** 鎖過期時間/毫秒，只對redis有效 */
    int timeoutMs() default 5000;

    /**
     * 方法引數field名稱，支援多級，如：方法引數名 或 方法引數名.物件名.物件名。
     * 利用反射取值，用於和key組合起來組成新的lockKey
     */
    String[] fields() default {};

    /** 獲取鎖失敗提示訊息，可將此訊息丟擲RuntimeException，然後用全域性異常處理器處理 */
    String failMsg() default "請勿重複提交|2101";

}

使用示例

1. 使用註解的方式，starter已配置AOP自動攔截帶有該註解的方法

@PostMapping("createOrder")
@Lockable(key = "order.addOder", waitTimeMs = 5000, timeoutMs = 5000, fields = {"product.id", "token"})
public RestResponse createOrder(@RequestBody Product product, @RequestParam(name = "token") String token) {
    // TODO createOrder
    return RestResponse.success();
}

@Transactional
@Lockable(key = "product.minusStock", waitTimeMs = 5000, timeoutMs = 5000, fields = "product.id")
public void minusStock(Product product) {
    // TODO 商品扣減庫存

}

1. 不使用註解，直接使用DistributeLock.tryLock和DistributeLock.releaseLock方法。注意釋放鎖程式碼必須要在獲得鎖的情況下才能執行，並且需要用try finally，如下：

@Transactional
public void minusStock(Product product) {
   // TODO 商品扣減庫存
   String lockValue = UUID.randomUUID().toString();
   boolean getLock = false;
   try {
       if (getLock = distributeLock.tryLock(lockKey, lockValue, waitTimeMs, timeoutMs)) {
           // TODO 獲取鎖成功，執行商品扣減庫存業務邏輯

       }
       // 獲取鎖失敗，執行失敗業務邏輯
       if (!getLock) {
           throw new RuntimeException("當前操作使用者過多，請稍後重試|2201");
       }
   } finally {
       // 獲取鎖成功才釋放鎖
       if (getLock) {
           distributeLock.releaseLock(lockKey, lockValue);
       }
   }
}

原文連結：https://my.oschina.net/dslcode/blog/2354141