概述

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

方案論證

這裏我們主要討論兩種方案：基於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

分布式鎖方案論證與實現