一、背景

在電商系統中，庫存的概念一定是有的，例如配一些商品的庫存，做商品秒殺活動等，而由於庫存操作頻繁且要求原子性操作，所以絕大多數電商系統都用Redis來實現庫存的加減，最近公司項目做架構升級，以微服務的形式做分布式部署，對庫存的操作也單獨封裝為一個微服務，這樣在高並發情況下，加減庫存時，就會出現超賣等問題，這時候就需要對庫存操作做分布式鎖處理。最近對分布式鎖的實現以及性能做了對比分析，今天記錄下來，與君共勉。

二、分布式鎖介紹

分布式鎖主要用於在分布式環境中保護跨進程、跨主機、跨網絡的共享資源實現互斥訪問，以達到保證數據的一致性。

分布式鎖要具有以下幾個特性：

1、可以保證在分布式部署的應用集群中，同一個方法在同一時間只能被一臺機器上的一個線程執行。

2、這把鎖要是一把可重入鎖（避免死鎖）

3、這把鎖最好是一把阻塞鎖（根據業務需求考慮要不要這條）

4、有高可用的獲取鎖和釋放鎖功能

5、獲取鎖和釋放鎖的性能要好

三、分布式鎖的幾種實現方式

1.基於zookeeper實現分布式鎖

2.采用中間件redisson提供分布式鎖

3.采用redis的watch做分布式鎖

4.采用redis的lua腳本編程方式實現分布式鎖

四、基於zookeeper實現分布式鎖的原理

1、zk的底層數據結構是樹形結構，由一個一個的數據節點組成；

2、zk的節點分為永久節點和臨時節點，客戶端可以創建臨時節點，當客戶端會話終止或超時後，zk會自動刪除臨時節點，該特性可以避免死鎖；

3、當節點的狀態發生變化時，zk的watch機制會通知監聽相應事件的客戶端，該特性可以可以用來實現阻塞等待加鎖；

4、客戶端可以在某個節點下創建子節點，Zookeeper會根據子節點數量自動生成整數序號，類似於數據庫的自增主鍵；

基於zk以上特性，可以實現分布式鎖，思路為：

創建一個永久節點作為鎖節點，試圖加鎖的客戶端在鎖節點下創建臨時順序節點。Zookeeper會保證子節點的有序性。若鎖節點下id最小的節點是為當前客戶端創建的節點，說明當前客戶端成功加鎖。否則加鎖失敗，訂閱上一個順序節點。當上一個節點被刪除時，當前節點為最小，說明加鎖成功。操作完成後，刪除鎖節點釋放鎖。

該方案的特征是優先排隊等待的客戶端會先獲得鎖，這種鎖稱為公平鎖。而鎖釋放後，所有客戶端重新競爭鎖的方案稱為非公平鎖。

五、代碼實現

1、引入相關zookeeper和curator相關jar

 1         <dependency>
 2             <groupId>org.apache.zookeeper</groupId>
 3             <artifactId>zookeeper</artifactId>
 4             <version>3.4.13</version>
 5             <scope>compile</scope>
 6             <exclusions>
 7                 <exclusion>
 8                     <groupId>org.slf4j</groupId>
 9                     <artifactId>slf4j-log4j12</artifactId>
10                 </exclusion>
11             </exclusions>
12         </dependency>
13         <dependency>
14             <groupId>org.apache.curator</groupId>
15             <artifactId>curator-recipes</artifactId>
16             <version>4.0.1</version>
17         </dependency>

2、curatorFramework初始化，放spring容器中

 1 /**
 2  * curatorFramework初始化
 3  *
 4  * @author LiJunJun
 5  * @date 2018/12/7
 6  */
 7 @Configuration
 8 public class CuratorBean {
 9 
10     @Bean
11     public CuratorFramework curatorFramework() {
12 
13         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
14         CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.10.110:2381", retryPolicy);
15         return client;
16     }
17 
18     @Bean
19     public InterProcessMutex interProcessMutex() {
20 
21         curatorFramework().start();
22 
23         return new InterProcessMutex(curatorFramework(), "/curator/lock");
24     }
25 }

3、業務代碼

 1     /**
 2      * interProcessMutex
 3      */
 4     @Resource
 5     private InterProcessMutex interProcessMutex;
 6 
 7     /**
 8      * 減庫存（基於zookeeper分布式鎖實現)
 9      *
10      * @param trace 請求流水
11      * @param stockManageReq（stockId、decrNum）
12      * @return -1為失敗，大於-1的正整數為減後的庫存量，-2為庫存不足無法減庫存
13      */
14     @Override
15     @ApiOperation(value = "減庫存", notes = "減庫存")
16     @RequestMapping(value = "/decrByStock", method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE, produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
17     public int decrByStock(@RequestHeader(name = "Trace") String trace, @RequestBody StockManageReq stockManageReq) {
18 
19         long startTime = System.nanoTime();
20 
21         LOGGER.reqPrint(Log.CACHE_SIGN, Log.CACHE_REQUEST, trace, "decrByStock", JSON.toJSONString(stockManageReq));
22 
23         int res = 0;
24         String stockId = stockManageReq.getStockId();
25         Integer decrNum = stockManageReq.getDecrNum();
26 
27         // 添加分布式鎖
28         boolean lockResult = false;
29 
30         try {
31             if (null != stockId && null != decrNum) {
32 
33                 stockId = PREFIX + stockId;
34 
35                 // 獲取鎖最多等待5s
36                 lockResult = interProcessMutex.acquire(5, TimeUnit.SECONDS);
37 
38                 if (!lockResult) {
39                     LOGGER.info("本次請求獲取鎖失敗，lockResult=1");
40                     return -1;
41                 }
42 
43                 // redis 減庫存邏輯
44                 String vStock = redisStockPool.get(stockId);
45 
46                 long realV = 0L;
47                 if (StringUtils.isNotEmpty(vStock)) {
48                     realV = Long.parseLong(vStock);
49                 }
50                 //庫存數  大於等於 要減的數目，則執行減庫存
51                 if (realV >= decrNum) {
52                     Long v = redisStockPool.decrBy(stockId, decrNum);
53                     res = v.intValue();
54                 } else {
55                     res = -2;
56                 }
57             }
58         } catch (Exception e) {
59             LOGGER.error(trace, "decr sku stock failure.", e);
60             res = -1;
61         } finally {
62             if (lockResult) {
63                 try {
64                     // 釋放鎖
65                     interProcessMutex.release();
66                 } catch (Exception e) {
67                     e.printStackTrace();
68                 }
69             }
70             LOGGER.respPrint(Log.CACHE_SIGN, Log.CACHE_RESPONSE, trace, "decrByStock", System.nanoTime() - startTime, String.valueOf(res));
71         }
72         return res;
73     }

六、ab壓測結果分析

發現性能低的簡直無法忍受，5000個請求，100並發量，tps僅有19.76，還有922個請求失敗

統計日誌中打印的獲取鎖失敗的請求個數，發現等待5s後仍未獲取到鎖數目就是就是ab壓測中失敗的922個

壓測過程中，我們可以看下zk的/curator/lock節點下的臨時節點變化情況，我們連接zk客戶端

./zkCli.sh -server 192.168.10.110:2381

查看目錄節點

ls /curator/lock，發現/curator/lock創建了很多臨時節點，並且隨著請求的執行，臨時節點也在不停的變化

七、總結

zookeeper確實可以實現分布式鎖，但由於需要頻繁的新增和刪除節點，性能比較差，不推薦使用。

下一篇我們分享基於redisson中間件實現的分布式鎖。

【原創】redis庫存操作，分布式鎖的四種實現方式[連載一]--基於zookeeper實現分布式鎖