分布式锁的应用场景与分布式锁实现（三）：基于Zookeeper实现分布式锁

分布式锁的应用场景与分布式锁实现（二）：基于Redis实现分布式锁

基于Zookeeper实现分布式锁

实现分布式锁目前有三种流行方案，分别为基于数据库、Redis、Zookeeper的方案。这里主要介绍基于zk怎么实现分布式锁。在实现分布式锁之前，先了解Zookeeper的相关知识。

Zookeeper相关知识

安装启动

安装：将zk安装包上传到/opt目录下，并切换到/opt目录，执行以下指令

# 解压
tar -zxvf zookeeper-3.7.0-bin.tar.gz
# 重命名
mv apache-zookeeper-3.7.0-bin/ zookeeper
# 打开zookeeper根目录
cd /opt/zookeeper
# 创建一个数据目录，备用
mkdir data
# 打开zk的配置目录
cd /opt/zookeeper/conf
# copy配置文件，zk启动时会加载zoo.cfg文件
cp zoo_sample.cfg zoo.cfg
# 编辑配置文件
vim zoo.cfg
# 修改dataDir参数为之前创建的数据目录：/opt/zookeeper/data
# 切换到bin目录
cd /opt/zookeeper/bin
# 启动 
./zkServer.sh start
./zkServer.sh status # 查看启动状态
./zkServer.sh stop # 停止
./zkServer.sh restart # 重启
./zkCli.sh # 查看zk客户端

如下，说明启动成功：

Java客户端

Zookeeper的Java客户端有：原生客户端、ZkClient、Curator框架(类似于Redisson，有很多功能性封装)。

引入原生客户端依赖

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.7.0</version>
</dependency>

常用API及其方法

package tech.msop.distributed.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Zookeeper测试类
 */
@Slf4j
public class ZkTest {

    public static void main(String[] args) throws InterruptedException {
        ZooKeeper zooKeeper = null;
        CountDownLatch countDownLatch = new CountDownLatch(1);
        try {
            zooKeeper = new ZooKeeper("192.168.56.99:2181", 30000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    Event.KeeperState state = watchedEvent.getState();
                    if (Event.KeeperState.SyncConnected.equals(state)
                            && Event.EventType.None.equals(watchedEvent.getType())
                    ) {
                        log.info("获取链接");
                        countDownLatch.countDown();
                    } else if (Event.KeeperState.Closed.equals(state)) {
                        log.info("关闭连接");
                    } else {
                        log.info("节点监测");
                    }

                }
            });
            countDownLatch.await();
            log.info("操作数据");
            // 节点新增：永久 临时 永久序列化 临时序列化
            // 创建一个节点，1-节点路径 2-节点内容 3-节点的访问权限 4-节点类型
            // OPEN_ACL_UNSAFE：任何人可以操作该节点
            // CREATOR_ALL_ACL：创建者拥有所有访问权限
            // READ_ACL_UNSAFE: 任何人都可以读取该节点
            // 创建永久节点
//            zooKeeper.create("/zookeeper/test","hello Zookeeper".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            // 创建临时节点
//            zooKeeper.create("/zookeeper/test1","hello Zookeeper".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            // 创建永久序列化节点
            zooKeeper.create("/zookeeper/test2", "hello Zookeeper".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
            // 创建临时序列化节点
//            zooKeeper.create("/zookeeper/test3","hello Zookeeper".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            log.info("节点查询");
            // 查询  判断节点是否存在 stat  节点监听,watch:true
            Stat stat = zooKeeper.exists("/zookeeper", true);
            if (stat != null) {
                log.info("当前节点存在");
            } else {
                log.warn("当前节点不存在");
            }
            // 获取当前节点中的数据内容 get
            byte[] data = zooKeeper.getData("/zookeeper", false, stat);
            log.info("当前节点的内容：{}", new String(data));
            // 获取当前节点的子节点 ls  节点监听,watch:true
            List<String> children = zooKeeper.getChildren("/zookeeper", new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    log.info("节点的子节点发生变化");
                }
            });
            log.info("当前节点的子节点：{}", children);

            // 更新：版本号必须和当前节点的版本号一致，否则更新失败。也可以指定为-1，表示不关心版本号
            zooKeeper.setData("/zookeeper", "hello zookeeper node".getBytes(), stat.getVersion());
            // 删除节点
//            zooKeeper.delete("/zookeeper/test1",-1);

            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            throw new RuntimeException(e);
        } finally {
            if (zooKeeper != null) {
                zooKeeper.close();
            }
        }

    }
}

思路分析

分布式锁的步骤：

1、获取锁：create一个节点
2、删除锁：delete一个节点
3、重试：没有获取到锁的请求重试

参照Redis分布式锁的特点：

1、互斥排他
2、防死锁：
- 1）可自动释放锁（临时节点）：获得锁之后客户端所在及其宕机了，客户端没有主动删除子节点；如果创建的是永久节点，那么这个锁永远不会释放，导致死锁；由于创建的是临时节点，客户端宕机后，过了一段时间Zookeeper没有收到客户端的心跳包判断会话失效，将临时节点删除从而释放锁。
- 2）可重入锁：借助ThreadLocal
3、防误删：宕机自动释放临时节点，不需要设置过期时间，也就不存在误删问题。
4、加锁/解锁要具备原子性
5、单点问题：使用Zookeeper可以有效的解决单点问题，ZK一般是集群部署的。
6、集群问题：Zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务。

基本实现

实现思路：

1、多个请求同时添加一个相同的临时节点，只有一个可以添加成功。添加成功的获取到锁
2、执行业务逻辑
3、完成业务流程后，删除节点释放锁。

由于Zookeeper获取连接是一个耗时过程，这里可以在项目启动时，初始化连接，并且只初始化一次。借助Spring特性，实现如下代码：

新增Zookeeper相关常量：

package tech.msop.distributed.lock.constants;

public interface StockConstant {
    /**
     * 默认库存数
     */
    Integer DEFAULT_STOCK_COUNT = 5000;
    /**
     * Redis 锁默认过期时间
     */
    long DEFAULT_REDIS_EXPIRE = 30;
    /**
     * Zookeeper分布式锁的连接地址
     */
    String ZOOKEEPER_LOCK_CONNECT_URL = "192.168.56.99:2181";
    /**
     * Zookeeper分布式锁的节点根路径
     */
    String ZOOKEEPER_LOCK_ROOT_PATH = "/distributed";
}

新增ZkClient客户端：

package tech.msop.distributed.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.CountDownLatch;

import static tech.msop.distributed.lock.constants.StockConstant.ZOOKEEPER_LOCK_CONNECT_URL;
import static tech.msop.distributed.lock.constants.StockConstant.ZOOKEEPER_LOCK_ROOT_PATH;

@Component
@Slf4j
public class ZkClient {

    private ZooKeeper zooKeeper = null;

    @PostConstruct
    public void init() {
        try {
            zooKeeper = new ZooKeeper(ZOOKEEPER_LOCK_CONNECT_URL, 30000, new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    log.info("获取链接成功");
                }
            });
            // 创建分布式锁根节点
            if (this.zooKeeper.exists(ZOOKEEPER_LOCK_ROOT_PATH, false) == null) {
                this.zooKeeper.create(ZOOKEEPER_LOCK_ROOT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 在Spring容器销毁时关闭连接
     */
    @PreDestroy
    public void destory() {
        // 释放zk的链接
        try {
            if (zooKeeper != null) {
                zooKeeper.close();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取ZK分布式锁
     *
     * @param lockName 锁名称
     * @return ZK分布式锁
     */
    public DistributedZkLock getLock(String lockName) {
        return new DistributedZkLock(zooKeeper, lockName);
    }
}

实现Zookeeper分布式锁：

package tech.msop.distributed.lock.zk;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.jetbrains.annotations.NotNull;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import static tech.msop.distributed.lock.constants.StockConstant.ZOOKEEPER_LOCK_ROOT_PATH;

public class DistributedZkLock implements Lock {

    private ZooKeeper zooKeeper;
    private String path;

    public DistributedZkLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.path  = ZOOKEEPER_LOCK_ROOT_PATH + "/" + lockName;
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        // 创建znode节点的过程：为了防止ZK客户端程序获取到锁之后，服务器宕机带来的死锁问题，这里创建的是临时节点
        try {
            this.zooKeeper.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            e.printStackTrace();
            try {
                Thread.sleep(80);
                this.tryLock();
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        // 删除zonde接地那的过程
        try {
            this.zooKeeper.delete(path, -1);
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }

    @NotNull
    @Override
    public Condition newCondition() {
        return null;
    }
}

改造服务，使用ZK分布式锁：

package tech.msop.distributed.lock.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import tech.msop.distributed.lock.constants.StockConstant;
import tech.msop.distributed.lock.entity.StockEntity;
import tech.msop.distributed.lock.lock.DistributedLockClient;
import tech.msop.distributed.lock.lock.DistributedRedisLock;
import tech.msop.distributed.lock.zk.DistributedZkLock;
import tech.msop.distributed.lock.mapper.StockMapper;
import tech.msop.distributed.lock.service.IStockService;
import tech.msop.distributed.lock.zk.ZkClient;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 库存服务实现类 <br/>
 */
@Service
@Slf4j
public class StockServiceImpl extends ServiceImpl<StockMapper, StockEntity>
        implements IStockService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private DistributedLockClient distributedLockClient;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ZkClient zkClient;

    /**
     * 减库存
     */
    @Override
    public void checkAndLock() {
        DistributedZkLock lock = zkClient.getLock("lock");
        lock.lock();
        try {
            // 1. 查询库存信息
            String stock = redisTemplate.opsForValue().get("stock").toString();
            // 2. 判断库存是否充足
            if (stock != null && stock.length() != 0) {
                Integer st = Integer.valueOf(stock);
                if (st > 0) {
                    // 3.扣减库存
                    redisTemplate.opsForValue().set("stock", String.valueOf(--st));
                }
            }
        } finally {
            lock.unlock();
        }
    }
}

使用Jmeter进行压力测试并查询库存余量：0（注意：所有测试之前都要重置库存量）

基本实现存在的问题：

性能一般（比MySQL分布式锁略好）
不可重入

优化：性能优化

基本实现中由于无限自旋影响性能：

试想：每个请求要想正常的执行完成，最终都是要创建节点，如果能够避免争抢必然可以提高性能。

这里借助于ZK的临时序列化节点，实现分布式锁：

1607048783043

监听实现阻塞锁

在设置事件监听时，每个客户端应该对刚好在它之前的子节点设置事件监听，序号为1的客户端监听序号为0的子节点删除消息，序号为2的监听序号为1的子节点删除消息。

所以调整后的分布式锁算法流程如下：

客户端连接Zookeeper，并在/distributed下创建临时的且有序的子节点，第一个客户端对应的子节点为/distributed/lock-000000000，第二个为/distributed/lock-000000001，以此类推。
客户端获取/lock下的子节点列表，判断自己创建的子节点是否为当前子节点列表中序号最小的子节点，如果是则认为获取锁，否则监听刚好在自己之前一位的子节点删除消息，获得子节点变更通知后重复此步骤直至获得锁；
执行业务代码；
完成业务流程后，删除对应的子节点释放锁。

改造分布式锁：

package tech.msop.distributed.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.jetbrains.annotations.NotNull;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;

import static tech.msop.distributed.lock.constants.StockConstant.ZOOKEEPER_LOCK_ROOT_PATH;

@Slf4j
public class DistributedZkLock implements Lock {

    private ZooKeeper zooKeeper;
    private String lockName;

    private String currentNodePath;

    public DistributedZkLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.lockName = lockName;
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        // 创建znode节点的过程：为了防止ZK客户端程序获取到锁之后，服务器宕机带来的死锁问题，这里创建的是临时节点
        try {
            currentNodePath = this.zooKeeper.create(ZOOKEEPER_LOCK_ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 获取前置节点，如果前置节点为空，则获取锁成功，否则监听前置节点
            String preNode = this.getPreNode();
            if (preNode != null) {
                // 利用闭锁思想，实现阻塞功能
                CountDownLatch countDownLatch = new CountDownLatch(1);
                // 因为获取前置节点这个操作，不具备原子性，再次判断ZK中的前置节点是否存在
                if (this.zooKeeper.exists(ZOOKEEPER_LOCK_ROOT_PATH + "/" + preNode, new Watcher() {

                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        countDownLatch.countDown();
                    }
                }) == null) {
                    return true;
                }
                countDownLatch.await();
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            // 重试获取锁
//            try {
//                Thread.sleep(200);
//                this.tryLock();
//            } catch (InterruptedException exception) {
//                exception.printStackTrace();
//            }
        }
        return false;
    }

    /**
     * 获取前置节点
     *
     * @return 前置节点路径
     */
    private String getPreNode() {
        // 获取根节点下的所有节点
        try {
            List<String> children = this.zooKeeper.getChildren(ZOOKEEPER_LOCK_ROOT_PATH, false);
            if (CollectionUtils.isEmpty(children)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 获取和当前节点同一资源的锁
            List<String> nodes = children.stream().filter(node -> StringUtils.startsWithIgnoreCase(node, lockName + "-")).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(nodes)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 排序
            Collections.sort(nodes);
            // 获取当前节点的下标
            // 获取当前接地那的名称
            // 获取当前节点
            String currentNode = StringUtils.substringAfterLast(currentNodePath, "/");
            int index = Collections.binarySearch(nodes, currentNodePath);
            if (index < 0) {
                throw new IllegalMonitorStateException("非法操作");
            } else if (index > 0) {
                return nodes.get(index - 1);// 返回前置节点
            }
            return null;
            // 如果当前节点就是第一个节点，返回null
        } catch (Exception e) {
            e.printStackTrace();
            throw new IllegalMonitorStateException("非法操作");
        }
    }

    @Override
    public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        // 删除zonde节点的过程
        try {
            this.zooKeeper.delete(currentNodePath, -1);
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }

    @NotNull
    @Override
    public Condition newCondition() {
        return null;
    }
}

优化：可重入锁

有两种实现方式：

在节点的内容中记录服务器、线程以及重入信息
ThreadLocal：线程的局部变量，线程私有

以ThreadLocal为例

package tech.msop.distributed.lock.zk;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.*;
import org.jetbrains.annotations.NotNull;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.stream.Collectors;

import static tech.msop.distributed.lock.constants.StockConstant.ZOOKEEPER_LOCK_ROOT_PATH;

@Slf4j
public class DistributedZkLock implements Lock {

    private ZooKeeper zooKeeper;
    private String lockName;

    private String currentNodePath;

    private static final ThreadLocal<Integer> THREAD_LOCAL = new ThreadLocal<>();

    public DistributedZkLock(ZooKeeper zooKeeper, String lockName) {
        this.zooKeeper = zooKeeper;
        this.lockName = lockName;
    }

    @Override
    public void lock() {
        this.tryLock();
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        // 判断ThreadLocal中是否已经有锁，有锁直接重入(+1)
        Integer flag = THREAD_LOCAL.get();
        if (flag != null && flag > 0){
            THREAD_LOCAL.set(flag + 1);
            return true;
        }

        // 创建znode节点的过程：为了防止ZK客户端程序获取到锁之后，服务器宕机带来的死锁问题，这里创建的是临时节点
        try {
            currentNodePath = this.zooKeeper.create(ZOOKEEPER_LOCK_ROOT_PATH + "/" + lockName + "-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            // 获取前置节点，如果前置节点为空，则获取锁成功，否则监听前置节点
            String preNode = this.getPreNode();
            if (preNode != null) {
                // 利用闭锁思想，实现阻塞功能
                CountDownLatch countDownLatch = new CountDownLatch(1);
                // 因为获取前置节点这个操作，不具备原子性，再次判断ZK中的前置节点是否存在
                if (this.zooKeeper.exists(ZOOKEEPER_LOCK_ROOT_PATH + "/" + preNode, new Watcher() {

                    @Override
                    public void process(WatchedEvent watchedEvent) {
                        countDownLatch.countDown();
                    }
                }) == null) {
                    THREAD_LOCAL.set(1);
                    return true;
                }
                countDownLatch.await();
            }
            THREAD_LOCAL.set(1);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            // 重试获取锁
//            try {
//                Thread.sleep(200);
//                this.tryLock();
//            } catch (InterruptedException exception) {
//                exception.printStackTrace();
//            }
        }
        return false;
    }

    /**
     * 获取前置节点
     *
     * @return 前置节点路径
     */
    private String getPreNode() {
        // 获取根节点下的所有节点
        try {
            List<String> children = this.zooKeeper.getChildren(ZOOKEEPER_LOCK_ROOT_PATH, false);
            if (CollectionUtils.isEmpty(children)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 获取和当前节点同一资源的锁
            List<String> nodes = children.stream().filter(node -> StringUtils.startsWithIgnoreCase(node, lockName + "-")).collect(Collectors.toList());
            if (CollectionUtils.isEmpty(nodes)) {
                throw new IllegalMonitorStateException("非法操作");
            }
            // 排序
            Collections.sort(nodes);
            // 获取当前节点的下标
            // 获取当前接地那的名称
            // 获取当前节点
            String currentNode = StringUtils.substringAfterLast(currentNodePath, "/");
            int index = Collections.binarySearch(nodes, currentNodePath);
            if (index < 0) {
                throw new IllegalMonitorStateException("非法操作");
            } else if (index > 0) {
                return nodes.get(index - 1);// 返回前置节点
            }
            return null;
            // 如果当前节点就是第一个节点，返回null
        } catch (Exception e) {
            e.printStackTrace();
            throw new IllegalMonitorStateException("非法操作");
        }
    }

    @Override
    public boolean tryLock(long time, @NotNull TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            // 重入锁 - 1
            THREAD_LOCAL.set(THREAD_LOCAL.get() - 1);
            if (THREAD_LOCAL.get() == 0){
                // 删除zonde节点的过程
                this.zooKeeper.delete(currentNodePath, -1);
            }
        } catch (InterruptedException | KeeperException e) {
            e.printStackTrace();
        }
    }

    @NotNull
    @Override
    public Condition newCondition() {
        return null;
    }
}

ZK分布式锁小结

参照Redis分布式锁的特点：

互斥排他：ZK节点的不可重复性，以及序列化节点的有序性
防死锁：
- 可自动释放锁：临时节点
- 可重入锁：借助于ThreadLocal
防误删：给每一个请求线程创建一个唯一的序列化节点
加锁/解锁需要具备原子性
- 创建节点、删除节点、查询及监听具备原子性
自动续期：没有过期时间，不需要自动续期
单点问题：使用ZooKeeper可以有效的解决单点问题，ZK一般是集群部署的
集群问题：Zookeeper集群是强一致性的，只要集群中有半数以上的机器存活，就可以对外提供服务
公平锁：有序性节点

Curator中的分布式锁

Curator是netflix公司开源的一套zookeeper客户端，目前是Apache的顶级项目。与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator解决了很多Zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册watcher和NodeExistsException异常等。

通过查看官方文档，可以发现Curator主要解决了三类问题：

封装Zookeeper client与Zookeeper server之间的连接处理
提供了一套Fluent风格的操作API
提供Zookeeper各种应用场景（recipe，例如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等）的抽象封装，这些实现都遵循了zk的最佳实践，并考虑了各种极端情况。

Curator由一系列的模块构成，对于一般开发中而言，常用的是curator-framework和curator-recipes：

curator-framework：提供了常见的ZK相关的底层操作
curator-recipes：提供了一些ZK的典型应用场景的参考。比如分布式锁。

引入依赖：需要注意curator传递进来的依赖，需要和实际服务器端使用的版本相符，所以会排除curator的依赖，使用符合服务器版本的zookeeper依赖

<!-- curator -->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>4.3.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>4.3.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
        </exclusion>
    </exclusions>
</dependency>
 <!-- Zookeeper -->
<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.7.0</version>
    <exclusions>
        <exclusion>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
        </exclusion>
    </exclusions>
</dependency>

添加Curator客户端配置

package tech.msop.distributed.lock.config;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import tech.msop.distributed.lock.constants.StockConstant;

/**
 * Curator配置
 */
@Configuration
public class CuratorConfig {
    @Bean
    public CuratorFramework curatorFramework(){
        // 初始化一个重试策略。这里使用的指数补偿策略。初始间隔时间（毫秒） 重试册数
        RetryPolicy retry = new ExponentialBackoffRetry(10000, 3);
        // 初始化curator客户端
        CuratorFramework client = CuratorFrameworkFactory.newClient(StockConstant.ZOOKEEPER_LOCK_CONNECT_URL, retry);
        // 手动启动，否则很多方法或者功能不工作的
        client.start();
        return client;
    }
}

可重入锁 InterProcessMutex

Reentrant和JDK的ReentrantLock类似，意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。它是由类InterProcessMutex实现。

// 常用构造方法
public InterProcessMutex(CuratorFramework client,String path)
// 获取锁
public void acquire();
// 带超时时间的可重入锁
public boolean acquire(long time,TimeUnit unit);
// 释放锁
public void release();

改造Service测试方法

package tech.msop.distributed.lock.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import tech.msop.distributed.lock.constants.StockConstant;
import tech.msop.distributed.lock.entity.StockEntity;
import tech.msop.distributed.lock.lock.DistributedLockClient;
import tech.msop.distributed.lock.lock.DistributedRedisLock;
import tech.msop.distributed.lock.mapper.StockMapper;
import tech.msop.distributed.lock.service.IStockService;
import tech.msop.distributed.lock.zk.ZkClient;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 库存服务实现类 <br/>
 */
@Service
@Slf4j
public class StockServiceImpl extends ServiceImpl<StockMapper, StockEntity>
        implements IStockService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private DistributedLockClient distributedLockClient;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ZkClient zkClient;
    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 减库存
     */
    @Override
    public void checkAndLock() {
        InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "/curator/locks");
        try {
            mutex.acquire();
            // 1. 查询库存信息
            String stock = redisTemplate.opsForValue().get("stock").toString();
            // 2. 判断库存是否充足
            if (stock != null && stock.length() != 0) {
                Integer st = Integer.valueOf(stock);
                if (st > 0) {
                    // 3.扣减库存
                    redisTemplate.opsForValue().set("stock", String.valueOf(--st));
                }
            }
            // 测试可重入性需要同一个InterProcessMutex对象
//            this.testSubLock(mutex);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                mutex.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试InterProcessMutex的可重入性
     * @param mutex 分布式锁
     */
    public void testSubLock(InterProcessMutex mutex) {
        try {
            mutex.acquire();
            log.info("测试可重入锁");
            mutex.release();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

注意：如想重入，则需要使用同一个InterProcessMutex对象

使用Jmeter进行压力测试并查询库存余量：0

不可重入锁 InterProcessSemaphoreMutex

具体先：InterProcessSemaphoreMutex。与InterProcessMutex调用方法类似，区别在于该锁是不可重入的，在同一个线程不可重入。

// 常用构造方法
public InterProcessSemaphoreMutex(CuratorFramework client,String path)
// 获取锁
public void acquire();
// 带超时时间的可重入锁
public boolean acquire(long time,TimeUnit unit);
// 释放锁
public void release();

改造方法：

package tech.msop.distributed.lock.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.redisson.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import tech.msop.distributed.lock.constants.StockConstant;
import tech.msop.distributed.lock.entity.StockEntity;
import tech.msop.distributed.lock.lock.DistributedLockClient;
import tech.msop.distributed.lock.lock.DistributedRedisLock;
import tech.msop.distributed.lock.mapper.StockMapper;
import tech.msop.distributed.lock.service.IStockService;
import tech.msop.distributed.lock.zk.ZkClient;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 库存服务实现类 <br/>
 */
@Service
@Slf4j
public class StockServiceImpl extends ServiceImpl<StockMapper, StockEntity>
        implements IStockService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private DistributedLockClient distributedLockClient;
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ZkClient zkClient;
    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 减库存
     */
    @Override
    public void checkAndLock() {
        InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curatorFramework, "/curator/locks");
        try {
            mutex.acquire();
            // 1. 查询库存信息
            String stock = redisTemplate.opsForValue().get("stock").toString();
            // 2. 判断库存是否充足
            if (stock != null && stock.length() != 0) {
                Integer st = Integer.valueOf(stock);
                if (st > 0) {
                    // 3.扣减库存
                    redisTemplate.opsForValue().set("stock", String.valueOf(--st));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                mutex.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

可重入读写锁 InterProcessReadWriteLock

类似JDK的ReentrantReadWriteLock。一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁，但读写不能升级成写锁。主要实现类：InterProcessReadWriteLock：

// 常用构造方法
public ReentrantReadWriteLock(CuratorFramework client,String path)
// 获取读锁对象
InterProcessMutex readLock();
// 获取写锁对象
InterProcessMutex writeLock();

注意：

读读可以并发
读写不可以并发
写写不可以并发
写锁在释放之前会阻塞请求线程，读锁不会

/**
* 测试ZK读锁
*/
@Override
public void testZkReadLock() {
    try {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(curatorFramework, "/curator/rwLock");
        readWriteLock.readLock().acquire(10, TimeUnit.SECONDS);
        log.info("操作数据");
        //            readWriteLock.readLock().release();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
* 测试ZK写锁
*/
@Override
public void testZkWriteLock() {
    try {
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(curatorFramework, "/curator/rwLock");
        readWriteLock.writeLock().acquire(10, TimeUnit.SECONDS);
        log.info("操作数据");
        //            readWriteLock.writeLock().release();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

联锁 InterProcessMultiLock（了解）

Multi Shared Lock是一个锁的容器。当调用acquire，所有的锁都会被acquire，如果请求失败，所有的锁都会被release。同样调用release时所有的锁都被release(失败被忽略)。基本上，它就是组锁的代表，在它上面的请求释放操作都会传递给它包含的所有的锁。实现类InterProcessMultiLock：

// 构造函数需要包含的锁的集合，或者一组ZooKeeper的path
public InterProcessMultiLock(List<InterProcessLock> locks);
public InterProcessMultiLock(CuratorFramework client, List<String> paths);

// 获取锁
public void acquire();
public boolean acquire(long time, TimeUnit unit);

// 释放锁
public synchronized void release();

信号量 InterProcessSemaphoreV2

一个计数的信号量类似JDK的Semaphore。JDK中Semaphore维护的一组许可（permits），而Cubator中称之为租约(Lease)。注意，所有的实例必须使用相同的numberOfLease值。调用acquire会返回一个租约对象。客户端必须在finally中close这些租约对象，否则这些租约会丢失掉。但是，如果客户端session由于某种原因比如crash丢掉，那么这些客户端持有的租约会自动close，这样其他客户端可以继续使用这些租约。主要实现类：InterProcessSemaphoreV2

// 构造方法
public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases);

// 注意一次你可以请求多个租约，如果Semaphore当前的租约不够，则请求线程会被阻塞。
// 同时还提供了超时的重载方法
public Lease acquire();
public Collection<Lease> acquire(int qty);
public Lease acquire(long time, TimeUnit unit);
public Collection<Lease> acquire(int qty, long time, TimeUnit unit)

// 租约还可以通过下面的方式返还
public void returnAll(Collection<Lease> leases);
public void returnLease(Lease lease);

测试案例，复制原Redis的信号量，修改为ZK信号量

public void testSemaphore() {
    // 设置资源量，请求的线程数
    InterProcessSemaphoreV2 semaphoreV2 = new InterProcessSemaphoreV2(curatorFramework, "/curator/semaphore", 5);
    try {
        Lease lease = semaphoreV2.acquire();// 获取资源，获取资源成功的线程可以继续处理业务操作，否则会被阻塞
        redisTemplate.opsForList().rightPush("log", "获取到了资源，开始处理业务逻辑" + Thread.currentThread().getName());
        TimeUnit.SECONDS.sleep(new Random().nextInt(10));
        redisTemplate.opsForList().rightPush("log", "处理完业务逻辑，释放资源" + Thread.currentThread().getName());
        semaphoreV2.returnLease(lease);// 手动释放资源，后续请求线程可以获取该资源
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}

栅栏 barrier

DistributedBarrier构造函数中barrierPath参数用来确定一个栅栏，只要barrierPath参数相同(路径相同)就是同一个栅栏。通常情况下栅栏的使用如下：

1、主client设置一个栅栏
2、其他客户端就会调用waitOnBarrier()等待栅栏移除，程序处理线程阻塞
3、主client移除栅栏，其他客户端的处理程序就会同时继续运行。

DistributedBarrier类的主要方法如下：

setBarrier() - 设置栅栏
waitOnBarrier() - 等待栅栏移除
removeBarrier() - 移除栅栏

DistributedDoubleBarrier双栅栏，允许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时，进程开始计算，当计算完成时，离开栅栏。DistributedDoubleBarrier实现了双栅栏的功能。构造函数如下：

// client - the client
// barrierPath - path to use
// memberQty - the number of members in the barrier
public DistributedDoubleBarrier(CuratorFramework client, String barrierPath, int memberQty);

enter()、enter(long maxWait, TimeUnit unit) - 等待同时进入栅栏
leave()、leave(long maxWait, TimeUnit unit) - 等待同时离开栅栏

memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到所有的成员都调用了enter。当leave方法被调用时，它也阻塞调用线程，直到所有的成员都调用了leave。

注意：参数memberQty的值只是一个阈值，而不是一个限制值。当等待栅栏的数量大于或等于这个值栅栏就会打开！

与栅栏(DistributedBarrier)一样,双栅栏的barrierPath参数也是用来确定是否是同一个栅栏的，双栅栏的使用情况如下：

1、从多个客户端在同一个路径上创建双栅栏(DistributedDoubleBarrier),然后调用enter()方法，等待栅栏数量达到memberQty时就可以进入栅栏。
2、栅栏数量达到memberQty，多个客户端同时停止阻塞继续运行，直到执行leave()方法，等待memberQty个数量的栅栏同时阻塞到leave()方法中。
3、memberQty个数量的栅栏同时阻塞到leave()方法中，多个客户端的leave()方法停止阻塞，继续运行。

共享计数器

利用Zookeeper可以实现一个集群共享的计数器。只要使用相同的path就可以得到最新的计数器值，这是由Zookeeper的一致性保证的。Curator有两个计数器，一个使用int来计数，一个用long来计数。

SharedCount

共享计数器SharedCount相关方法如下：

// 构造方法
public SharedCount(CuratorFramework client, String path, int seedValue);
// 获取共享计数的值
public int getCount();
// 设置共享计数的值
public void setCount(int newCount) throws Exception;
// 当版本号没有变化时，才会更新共享变量的值
public boolean  trySetCount(VersionedValue<Integer> previous, int newCount);
// 通过监听器监听共享计数的变化
public void addListener(SharedCountListener listener);
public void addListener(final SharedCountListener listener, Executor executor);
// 共享计数在使用之前必须开启
public void start() throws Exception;
// 关闭共享计数
public void close() throws IOException;

使用案例：

public void testShareCount() {
    try {
        // 第三个参数是共享计数器的初始值
        SharedCount sharedCount = new SharedCount(curatorFramework, "/curator/sharecount", 100);
        // 启动共享计数器
        sharedCount.start();
        // 获取共享计数器的值
        int count = sharedCount.getCount();
        // 修改共享计数器的值
        int random = new Random().nextInt(100);
        sharedCount.setCount(random);
        log.info("共享计数器的初始值：{}，现在修改成了：{}",count,random);
    } catch (Exception e) {
        e.printStackTrace();
    }
}

DistributedAtomicNumber

DistributedAtomicNumber接口是分布式原子数值类型的抽象，定义了分布式原子数值类型需要提供的方法。

DistributedAtomicNumber接口有两个实现：DistributedAtomicLong 和 DistributedAtomicInteger

这两个实现将各种原子操作的执行委托给了DistributedAtomicValue，所以这两种实现是类似的，只不过表示的数值类型不同而已。

DistributedAtomicLong除了计数的范围比SharedCount大了之外，比SharedCount更简单易用。它首先尝试使用乐观锁的方式设置计数器，如果不成功(比如期间计数器已经被其它client更新了)，它使用InterProcessMutex方式来更新计数值。此计数器有一系列的操作：

get(): 获取当前值
increment()：加一
decrement(): 减一
add()：增加特定的值
subtract(): 减去特定的值
trySet(): 尝试设置计数值
forceSet(): 强制设置计数值

你必须检查返回结果的succeeded()，它代表此操作是否成功。如果操作成功， preValue()代表操作前的值， postValue()代表操作后的值。