【连载】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

        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.13</version>
            <scope>compile</scope>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>

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

/**
 * curatorFramework初始化
 *
 * @author LiJunJun
 * @date 2018/12/7
 */
@Configuration
public class CuratorBean {

    @Bean
    public CuratorFramework curatorFramework() {

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.10.110:2381", retryPolicy);
        return client;
    }

    @Bean
    public InterProcessMutex interProcessMutex() {

        curatorFramework().start();

        return new InterProcessMutex(curatorFramework(), "/curator/lock");
    }
}

3、业务代码

    /**
     * interProcessMutex
     */
    @Resource
    private InterProcessMutex interProcessMutex;

    /**
     * 减库存（基于zookeeper分布式锁实现)
     *
     * @param trace 请求流水
     * @param stockManageReq（stockId、decrNum）
     * @return -1为失败，大于-1的正整数为减后的库存量，-2为库存不足无法减库存
     */
    @Override
    @ApiOperation(value = "减库存", notes = "减库存")
    @RequestMapping(value = "/decrByStock", method = RequestMethod.POST, consumes = MediaType.APPLICATION_JSON_UTF8_VALUE, produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public int decrByStock(@RequestHeader(name = "Trace") String trace, @RequestBody StockManageReq stockManageReq) {

        long startTime = System.nanoTime();

        LOGGER.reqPrint(Log.CACHE_SIGN, Log.CACHE_REQUEST, trace, "decrByStock", JSON.toJSONString(stockManageReq));

        int res = 0;
        String stockId = stockManageReq.getStockId();
        Integer decrNum = stockManageReq.getDecrNum();

        // 添加分布式锁
        boolean lockResult = false;

        try {
            if (null != stockId && null != decrNum) {

                stockId = PREFIX + stockId;

                // 获取锁最多等待5s
                lockResult = interProcessMutex.acquire(5, TimeUnit.SECONDS);

                if (!lockResult) {
                    LOGGER.info("本次请求获取锁失败，lockResult=1");
                    return -1;
                }

                // redis 减库存逻辑
                String vStock = redisStockPool.get(stockId);

                long realV = 0L;
                if (StringUtils.isNotEmpty(vStock)) {
                    realV = Long.parseLong(vStock);
                }
                //库存数  大于等于 要减的数目，则执行减库存
                if (realV >= decrNum) {
                    Long v = redisStockPool.decrBy(stockId, decrNum);
                    res = v.intValue();
                } else {
                    res = -2;
                }
            }
        } catch (Exception e) {
            LOGGER.error(trace, "decr sku stock failure.", e);
            res = -1;
        } finally {
            if (lockResult) {
                try {
                    // 释放锁
                    interProcessMutex.release();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            LOGGER.respPrint(Log.CACHE_SIGN, Log.CACHE_RESPONSE, trace, "decrByStock", System.nanoTime() - startTime, String.valueOf(res));
        }
        return res;
    }

六、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中间件实现的分布式锁。