原创redis库存操作，分布式锁的四种实现方式[连载一]--基于zookeeper实现分布式锁

Posted 2021-01-28 李军军

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了原创redis库存操作，分布式锁的四种实现方式[连载一]--基于zookeeper实现分布式锁相关的知识，希望对你有一定的参考价值。

一、背景

在电商系统中，库存的概念一定是有的，例如配一些商品的库存，做商品秒杀活动等，而由于库存操作频繁且要求原子性操作，所以绝大多数电商系统都用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     }