??理论篇

一、基础概念

ZooKeeper是开源分布式协调服务，提供高可用、高性能、稳定的分布式数据一致性解决方案，通常被用于实现诸如数据发布/订阅、负载均衡、命名服务、分布式协调/通知、集群管理、Master选举、分布式锁和分布式队列等功能。

二、ZooKeeper数据模型

2.1 znode(数据节点)

Zookeeper中所有存储的数据由znode组成，节点也成为znode，并以key value键值对形式存储数据。整体结构类似linux文件系统，根路径以/开头。

znode中数据的读写都是原子的，而且每一个znode都有一个Access Control List(ACL)用来限制谁可以做什么。每一个znode的数据大小不能超过1M

2.1.1 znode数据节点名称规范

null 字符(即\u0000)不能组成znode节点path的命名
\ud800 - uF8FF, \uFFF0 - uFFFF， \u0001 - \u001F 以及 \u007F、\u009F 这些玩意无法很好地进行展示，看起来像乱码
"."可以构成命名的一部分，但是不能单独作为path的命名
"zookeeper"是保留字

2.1.2 znode数据节点组成：

stat 状态属性组成：

属性名称	属性描述
czxid	创建节点的事务ID
mzxid	节点最后一次修改的事务ID
pzxid	子节点列表最后的一次修改(子节点列表的增加或删除)的事务ID
ctime	创建节点的时间，单位：毫秒
mtime	节点最后一次修改的时间，单位：毫秒
version	节点数据变更的次数
cversion	子节点变更的次数
aversion	节点ACL变更的次数
ephemeralOwner	如果节点是临时节点，则此值为创建该节点的session的id，否则为0
dataLength	节点数据长度
numChildren	子节点个数

data
children

2.1.3 znode数据节点类型：

持久节点(Persistent Nodes)
临时节点(Ephemeral Nodes) 临时节点的生命周期即为创建这些节点的会话生命周期，即会话结束，则这些节点就会被删除。所以临时节点不允许创建子节点。
顺序节点(Sequence Nodes ) 顺序节点可以是持久的，也可以是临时的。在创建节点时，可为节点路径添加一个单调递增计数器，Zookeeper将通过将10位的序列号附加到原始节点名称后来设置节点路径。
容器节点(Container Nodes) 此类型节点是3.6.0版本之后添加，容器节点是一种有特殊用途的节点，可用于leader选举和分布式锁等，当容器中最后一个子节点被删除，此容器节点将会在未来某个时刻被删除。
超时过期节点(TTL Nodes) 此类型节点是3.6.0版本之后添加，当创建一个持久节点或者顺序持久节点时，可以为其设置一个毫秒级的超时过期时间，如果在设置时间内，此节点没又被修改过而且也没有子节点，则此节点将作为候选项，在未来某个时刻被删除。当然TTL Nodes默认是禁用状态的。

三、Zookeeper Time

Zookeeper中时间有很多表示时间的方式。

Zxid 事务ID。Zookeeper状态的每次变化都会收到这样一个事务ID
Version numbers
Ticks
Real time

四、ZooKeeper Sessions (会话)

Zookeeper客户端通过

五、ZooKeeper Watches (监听)

5.1 watch基本概念

Zookeeper所有读相关操作：getData()、getChildren()、 exists()等都有一个参数boolean watch用来设置watche。按照读的内容不同，有两种watch：Data Watch和Child Watch，像getData()和exists()这种属于读取znode数据，所以属于Data Watch，所以当znode数据发生变更，将触发znode的Data Watch；getChildren()对应的就是Child Watch。创建子节点将触发父节点的Child Watch，而节点的删除将同时触发Data Watch和Child Watch。

Watch是一个一次性触发器，比如当调用 getData("/znode1", true) 后--true代表设置监听，该节点被删除或者数据发生变更，将会触发客户端注册的watch,触发之后，将被删除，也就是往后数据再变化就不会再触发此watch了。

5.2 watch 事件类型

那么当触发watch的时候有因为数据发生变化的，有因为节点创建或删除的等等，客户端如何判断服务端触发的watch是各种类型呢？zookeeper提供了EventType的枚举，服务端触发watch时，会告诉客户端是何种类型。

一共有如下这么多事件类型：

None
NodeCreated
NodeDeleted
NodeDataChanged
NodeChildrenChanged
DataWatchRemoved
ChildWatchRemoved
PersistentWatchRemoved

其中最后三个事件类型：DataWatchRemoved、ChildWatchRemoved、PersistentWatchRemoved分别是删除不同类型的watch的时候事件类型，从单词字面意思也应该能够理解。

5.3 永久递归watch

3.6.0之后(包括3.6.0) 客户端可以通过addWatch()为znode设置永久、递归的watch，这意味着watch不再是一次性的了，可以多次触发不会被删除，并且还会递归触发。当然也有移除永久watch的机制：removeWatches()

watch是维护在zookeeper服务端的，所有当客户端与服务端断链，将不会接受到watch的触发，而当重连后都将恢复可以重新触发。

5.3 关于watch的一些注意事项?

1.标准watch是一次性的，如果当客户端接收到watch的回调通知，那么此watch将被删除，如果客户端还想接收到通知，则需要注册另一个watch

2.正如第一条所讲，在客户端接收到watch回调通知时，可能会继续设置一个watch以监听znode的下次变更，但是假如在接收到watch和发送请求设置新watch的中间，znode发生了多次变化，这个可能客户端会接收不到此变更通知。

3.如果为比如exist、getData注册了同一个watch,那么当watch被删除的时候，仅会触发一次delete watch事件

六、ZooKeeper access control using ACLs(权限控制器)

ACL全称Access Control List，即访问控制列表。Zookeeper的ACL实现很类似与UNIX的文件系统权限控制。每一个ACL针对指定的znode,但是并不针对指定znode的子节点，也就是ACL并不递归生效。假如给/app设置了一个ACL，那么/app/childtest将不受此ACL控制。

权限的表达式为scheme:id, permissions

6.1 scheme-->授权策略

授权策略一共有4种

授权策略	含义
world	默认策略。任何人都可以访问
auth	即已经认证通过的用户
digest	通过使用MD5进行哈希 username:password格式生成的字符串来进行身份验证，当进行身份验证时，是使用usename:password明文形式字符串，当用做ACL验证时，会先经过base64编码，然后使用SHA1加密
ip	使用客户端IP作为ACL身份标识。其格式为addr/bits，bits代表客户端的IP地址要至少匹配ACL中IP地址的多少位
x509

6.2 id-->授权对象

6.3 permissions-->权限点

权限点	含义
CREATE	可以创建子节点
READ	可以获取znode数据，以及znode的子节点列表
WRITE	可以为znode设置数据
DELETE	可以删除znode的子节点
ADMIN	可以为znode设置ACL，ADMIN就像是znode的owner

六、ZooKeeper Consistency Guarantees(一致性保障)

Zookeeper是高性能、可扩展的服务，它的读操作和写操作都非常的快

6.1 Sequential Consistency(顺序一致性)

来自客户端的更新将会按照顺序进行处理

6.2 Atomicity (原子性)

6.3 Single System Image(单一系统镜像)

无论客户端连接到哪一个服务器上，其看到的服务端数据模型都是一致的

6.4Reliability(可靠性)

一旦更新请求被处理，更改的结果将会持久化

6.5 Timeliness (及时性)

??实战篇

一、下载安装

1.1 下载

下载地址：https://zookeeper.apache.org/releases.html#download [apache-zookeeper-3.6.2-bin.tar.gz](https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.6.2/apache-zookeeper-3.6.2-bin.tar.gz)

1.2 单机模式

1.2.1 设置配置文件

在conf目录下会有一个zoo_sample.cfg文件，这里提供了样例配置信息，我们只需要将此文件改名为zoo.cfg，这样才会被zookeeper识别。该配置文件中的配置项说明：

tickTime 单位：毫秒。是服务器之间或客户端与服务器之间维持心跳的时间间隔；且最小会话超时时间将是tickTime的两倍
dataDir zookeeper保存的数据的目录地址，默认情况下，事务log也会记在这里(除非另外指定)
clientPort 客户端连接服务端的端口 zoo.cfg示例

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181

1.2.2 启动zookeeper服务端(前提是需要保证安装机器上有JDK)

./zkServer.sh start

1.2.3 客户端连接服务器

./zkCli.sh -server 127.0.0.1:2181

1.3 集群模式

zookeeper集群部署可以获得高可靠性，要想实现高可靠容错好集群，至少需要3台服务器，且集群数量最好为奇数。

1.3.1 设置集群配置文件zoo.conf

tickTime=2000
dataDir=/var/lib/zookeeper/
clientPort=2181
initLimit=5
syncLimit=2
server.1=zoo1:2888:3888
server.2=zoo2:2888:3888
server.3=zoo3:2888:3888

除了单机模式中需要配置的那几个参数外，需要配置和集群相关的参数

initLimit 表示集群中的followers服务器在连接leader服务器时，经过最大initLimit个tickTime后，若leader还未接收到followers的信息，则认为连接失败
syncLimit 表示集群中follower服务器与leader服务器在同步消息时最大syncLimit*tickTime时间间隔未收到响应，则此follower会被抛弃
server.id=host:port:port 配置文件中每行的server.id=host:port:port共同组成一个集群。这里有两个端口，前面的端口用于与集群leader连接的端口，后面的端口用于leader选举。如果想在同一台机器上搭建伪集群，则第一个端口不同即可

1.3.2 设置myid文件

myid文件中只有一行内容，且这行内容就是上述配置server.id=host:port:port 中的id，在集群模式下该id不可重复，范围为1-255，将此文件放在dataDir参数表示的目录下

1.3.3 查看服务器状态

./zkServer.sh status

1.3.4 搭建集群过程遇到的问题

在搭建集群的时候，启动三台机器都显示启动成功，但是使用客户端命令也连接失败，通过./zkServer.sh status 显示Error contacting service. It is probably not running.查看logs下的日志，发现有这么一行：Exception when following the leader java.io.EOFException。查资料发现是我将客户端端口和follower服务器与leader服务器通信的端口混到一起了，如下图配置的相同，所以出现了这种错误，所以这两个端口不可以相同

二、zk客户端操作 Zookeeper及基础命令

2.1 bin/zkServer.sh

用于操作zk服务器相关，不同的参数代表不同的操作

#启动zk服务器
bin/zkServer.sh start

#重启zk服务器
bin/zkServer.sh restart

#停止zk服务器
bin/zkServer.sh stop

#查看zk运行状态
bin/zkServer.sh status

#查看zk版本信息
bin/zkServer.sh version

2.2 bin/zkCli.sh

启动客户端连接zk服务器

bin/zkCli.sh -server {host}:{port}

2.3 客户端命令

ls列出指定路径下所有子节点 ls [-s] [-w] [-R] path

#列出根路径下所有节点
ls /
#列出指定路径节点下的子节点同时，输出指定路径节点状态细腻些
ls -s /

get查看指定路径节点信息 get [-s] [-w] path

#查看/node1节点信息
get /node1
# 查看/node1节点信息及状态信息
get -s /node1

create创建节点 create [-s] [-e] [-c] [-t ttl] path [data] [acl] 可选参数 -s 代表创建顺序节点可选参数 -e 代表创建临时节点可选参数 -c 代表创建容器节点可选参数 -t 设置节点过期时间

# 创建路径为/node1，数据为data的持久节点
create /node1 data

set修改节点 set [-s] [-v version] path data 可选参数-s代表更新节点后，输出节点状态信息可选参数-v用来表示根据版本号进行更新，低版本肯定是无法更新高版本节点的(乐观锁)

#创建/node2节点
[zk: localhost:2181(CONNECTED) 27] create /node2 data
Created /node2
#查看/node2节点状态信息
[zk: localhost:2181(CONNECTED) 28] get -s /node2
data
#...省略其他信息
dataVersion = 0
#...省略其他信息
#更新节点，此时版本号为1
[zk: localhost:2181(CONNECTED) 29] set -s /node2 data2
#...省略其他信息
dataVersion = 1
#...省略其他信息
[zk: localhost:2181(CONNECTED) 30] set -v 1 /node2 data3
#更新失败
[zk: localhost:2181(CONNECTED) 31] set -v 1 /node2 data3
version No is not valid : /node2
[zk: localhost:2181(CONNECTED) 32]

deletedelete [-v version] path 删除节点因为删除本身也是更新的意思，-v参数同上set的-v参数

#删除/node2节点
delete /node2

三、Java 操作 Zookeeper

通过Java操作Zookeeper有两种方式，一种就是通过Zookeeper提供的原生API(链接)进行操作，一种就是通过Apache Curator(官网) ---进行操作。

3.1 Apache Curator是什么

Apache Curator是比较完善的Zookeeper客户端框架，针对Zookeeper原生API的封装和扩展，降低了使用 Zookeeper的复杂性，使得使用Zookeeper更加可靠、更加简单。Curator有很多不同的artifacts，可以根据我们的需要进行引入使用：

curator-recipes 该artifact包含了对Zookeeper的所有操作，大部分场景只需要使用该artifact即可满足需求。
curator-framework 针对zookeeper的高级功能的简化封装，该artifact构建在整个客户端之上，所有应该自动包含此模块
curator-client 对于Zookeeper客户端链接相关操作的封装

3.2 使用apache-curator操作zookeeper

3.2.1 引入pom

org.apache.curatorcurator-recipes5.1.0

5.1.0版本对应的zookeeper客户端版本为3.6.0

3.2.2 创建连接

创建链接需要zk host地址，需要重试策略，还可以设置一些诸如超时时间等等参数。创建客户端连接的入口类是CuratorFrameworkFactory，该类中有一个内部静态类Builder，用于设置连接的一些额外高级参数。重试策略则是RetryPolicy。

    public static CuratorFramework createWithOptions(String connectionString, RetryPolicy retryPolicy, int connectionTimeoutMs, int sessionTimeoutMs){
        return CuratorFrameworkFactory.builder()
                .connectString(connectionString)
                .retryPolicy(retryPolicy)
                .connectionTimeoutMs(connectionTimeoutMs)
                .sessionTimeoutMs(sessionTimeoutMs)
                .build();
    }

3.2.3 创建节点，更新节点内容

    /**
     * 创建持久性节点
     * @param client CuratorFramework
     * @param path 节点路径
     * @param payload 节点内容
     * @throws Exception
     */
    public static void create(CuratorFramework client, String path, byte[] payload) throws Exception {
        client.create().forPath(path, payload);
    }

更多参考官方示例：https://github.com/apache/curator/tree/master/curator-examples/src/main/java/framework

??应用篇

一、应用场景

命名服务：按名称标识集群中的节点
统一配置管理
数据发布/订阅
分布式锁
Leader 选举

二、通过Zookeeper实现统一配置管理

三、通过Zookeeper实现分布式锁

Apache Curator针对分布式锁提供了多种实现

InterProcessMutex：分布式可重入排它锁
InterProcessSemaphoreMutex：分布式排它锁
InterProcessReadWriteLock：分布式可重入读写锁
InterProcessMultiLock：将多个锁作为单个实体管理的容器

3.1 代码实战

3.1.1 分布式可重入排它锁

/**
 * @author miaomiao
 * @date 2020/10/25 11:15
 */
public class DistributReetrantLock {

    private final InterProcessMutex interProcessMutex;
    private final String lockPath;

    public DistributReetrantLock(CuratorFramework client, String lockPath) {
        this.lockPath = lockPath;
        // 此InterProcessMutex构造方法的maxLeases为1，表示为排他锁
        this.interProcessMutex = new InterProcessMutex(client, lockPath);
    }

    /**
     * 阻塞式获取
     */
    public void tryLock() throws Exception {
        this.interProcessMutex.acquire();
    }

    /**
     * 超时未获取到锁则获取锁失败
     * @param time
     * @param unit
     * @return 是否获取到锁
     * @throws Exception
     */
    public boolean tryLock(long time, TimeUnit unit) throws Exception {
        return this.interProcessMutex.acquire(time,unit);
    }

    /**
     * 释放锁
     * @throws Exception
     */
    public void unLock() throws Exception {
        this.interProcessMutex.release();
    }
}

测试

     public static void main(String[] args) throws InterruptedException {

        ExecutorService executorService = Executors.newFixedThreadPool(5);

        final String lockPath = "/lock";

        for (int i = 0; i             final int clientIndex = i;
            Callable callable = new Callable() {
                public Void call() throws Exception {
                    CuratorFramework simpleClient = MyZookeeperClient.createSimpleClient("192.168.0.104:2181");
                    try {
                        simpleClient.start();
                        DistributReetrantLock distributReetrantLock = new DistributReetrantLock(simpleClient, lockPath);
                        // 阻塞式获取
                        distributReetrantLock.tryLock();
                        System.out.println("Client:" + clientIndex + " get lock!");
                        // 验证是否是可重入的
                        distributReetrantLock.tryLock();
                        System.out.println("Client:" + clientIndex + " get lock again!");
                        Thread.sleep(1000);
                        // 持有锁一秒后释放，以便其他客户端获取到锁
                        System.out.println("Client:" + clientIndex + " release lock!");
                        distributReetrantLock.unLock();
                    } finally {
                        CloseableUtils.closeQuietly(simpleClient);
                    }return null;
                }
            };
            executorService.submit(callable);
        }
        executorService.awaitTermination(10, TimeUnit.MINUTES);
    }

输出结果

Client:4 get lock!
Client:4 get lock again!
Client:4 release lock!
Client:3 get lock!
Client:3 get lock again!
Client:3 release lock!
Client:0 get lock!
Client:0 get lock again!
Client:0 release lock!
Client:1 get lock!
Client:1 get lock again!
Client:1 release lock!
Client:2 get lock!
Client:2 get lock again!
Client:2 release lock!

3.1.2 分布式排它锁

/**
 * 分布式排它锁
 * @author miaomiao
 * @date 2020/10/25 12:54
 */
public class DistributeLock {
    private final String lockPath;
    private InterProcessSemaphoreMutex interProcessSemaphoreMutex;
    public DistributeLock(CuratorFramework client,String lockPath){
        this.lockPath = lockPath;
        this.interProcessSemaphoreMutex = new InterProcessSemaphoreMutex(client,lockPath);
    }
    /**
     * 阻塞式获取
     */
    public void tryLock() throws Exception {
        this.interProcessSemaphoreMutex.acquire();
    }

    /**
     * 超时未获取到锁则获取锁失败
     * @param time
     * @param unit
     * @return 是否获取到锁
     * @throws Exception
     */
    public boolean tryLock(long time, TimeUnit unit) throws Exception {
        return this.interProcessSemaphoreMutex.acquire(time,unit);
    }

    /**
     * 释放锁
     * @throws Exception
     */
    public void unLock() throws Exception {
        this.interProcessSemaphoreMutex.release();
    }
}

3.1.3 分布式可重入读写锁

获取到写锁的进程可以继续获取读锁，当释放掉写锁后，降级为读锁。

/**
 * 分布式可重入读写锁
 * @author miaomiao
 * @date 2020/10/25 13:07
 */
public class DistributeReetrantReadWriteLock {
    private InterProcessReadWriteLock interProcessReadWriteLock;
    private String lockPath;
    public DistributeReetrantReadWriteLock(CuratorFramework client,String lockPath){
        this.lockPath = lockPath;
        this.interProcessReadWriteLock = new InterProcessReadWriteLock(client,lockPath);
    }

    /**
     * 阻塞式获取读锁
     * @throws Exception
     */
    public void tryReadLock() throws Exception {
       interProcessReadWriteLock.readLock().acquire();
    }

    /**
     * 获阻塞式获取写锁
     * @throws Exception
     */
    public void tryWriteLock() throws Exception {
        interProcessReadWriteLock.writeLock().acquire();
    }

    /**
     * 释放写锁
     * @throws Exception
     */
    public void unlockWriteLock() throws Exception {
        interProcessReadWriteLock.writeLock().release();
    }

    /**
     * 释放读锁
     * @throws Exception
     */
    public void unlockReadLock() throws Exception {
        interProcessReadWriteLock.readLock().release();
    }
}

3.2 分布式锁原理

3.1 排它锁原理

利用 zookeeper 的同级节点的唯一性特性，在需要获取排他锁时，所有的客户端试图通过调用 create() 接口，在 /指定节点下创建相同临时子节点 /exclusive_lock/lock，最终只有一个客户端能创建成功，那么此客户端就获得了分布式锁。同时，所有没有获取到锁的客户端可以在 /exclusive_lock 节点上注册一个子节点变更的 watcher 监听事件，以便重新争取获得锁。

3.2 读写锁原理

共享锁需要实现共享读，排他写。实现原理：当多个客户端请求共享锁时，为指定节点创建临时顺序子节点，且子节点的path能够区分是哪个客户端以及当前该客户端的操作(写还是读)就像这样 [hostname]-请求类型W/R-序号。之后在判断当前客户端是否获得锁时，如果当前客户端的操作为读请求，则判断如果存在小于自己节点序号的写请求节点或者自己本身就是最小序列的节点，则获取到锁；如果当前客户端的操作为写请求时，则只有自己节点序号是最小的节点时，才可以获取到锁。如果没有获取到锁，读请求在比自己序号小的最后一个写请求节点添加监听器；写请求在子节点列表比自己小的最后一个节点注册watcher监听。

四、通过Zookeeper实现Leader选举

在分布式系统中，leader选举是指指定一个进程(一个实例、一台机器)作为分配给多台服务器任务的组织者的过程。在任务开始之前，所有服务器节点都不知道哪个节点将作为任务的领导者或者说协调者，然后在leader选举之后，每个节点都会识别出一个特定的、唯一的节点作为任务leader。

Apache Curator针对Leader 选举提供了两种方式：

4.1 利用顺序临时节点实现

最简单的方式就是，当有一个"/election"节点，客户端们为此节点创建一个顺序、临时节点，每个客户端创建的子节点都会自动带上一个序号后缀，并且最早创建的序号最小，只需要选举序号最小的子节点对应的客户端作为leader即可。

当然这些肯定是远远不够的，还要有假如leader宕机出现故障，必须要重新推举新的leader机制。一种解决办法就是所有的应用客户端都监听序号最小的子节点来判断自己是否可以成为leader,因为假如leader客户端宕机，那么最小序号节点也会消失，所以会产生新的最小序号节点，也就是产生新的leader。但是这样做会产生羊群效应(herd effect)：所有的客户端都接收到了最小序号子节点被删除的通知，接下来所有客户端都调用getChilrden()获取"/election"的子节点列表，如果客户端数量很大，将会给zookeeper服务器带来一定的压力。为了避免羊群效应，每个客户端只需要监听自己对应子节点的前一个节点就足够了，这样当leader客户端宕机，最小子序列节点被删除，那么最小序列子节点的下一个节点对应的客户端就成为新的leader。

对应在Apache Curator中的相关实现类为

org.apache.curator.framework.recipes.leader.LeaderLatch

核心类，主入口

org.apache.curator.framework.recipes.leader.LeaderLatchListener leader latch监听器，当leader状态发生改变时回调，该接口有两个方法

  //当称为leader时调用
  public void isLeader();
  //当失去leader时调用
  public void notLeader();

示例

                        CuratorFramework simpleClient = MyZookeeperClient.createSimpleClient("192.168.0.104:2181");

                        simpleClient.start();

                        MyLeaderLatch leaderLatch = new MyLeaderLatch(simpleClient,"/leader_election" ,new LeaderLatchListener(){
                            public void isLeader() {
                                System.out.println("Client:"+clientIndex+" is leader");
                            }

                            public void notLeader() {
                                System.out.println("Client:"+clientIndex+" lose leader");

                            }
                        });
                        leaderLatch.start();

4.2 利用分布式锁实现

??参考文章

https://github.com/Snailclimb/JavaGuide/blob/master/docs/system-design/framework/zookeeper/

https://zookeeper.apache.org/doc/r3.6.2/zookeeperProgrammers.html

https://www.cnblogs.com/qlqwjy/p/10517231.html

ZooKeeper 的应用场景

分布式服务框架 Zookeeper —— 管理分布式环境中的数据

https://www.runoob.com/w3cnote_genre/zookeeper/page/2

http://curator.apache.org/