1 ZooKeeper系统模型

ZooKeeper数据模型Znode

在ZooKeeper中，数据信息被保存在⼀个个数据节点上，这些节点被称为znode。ZNode 是Zookeeper 中最⼩数据单位，在 ZNode 下⾯⼜可以再挂 ZNode，这样⼀层层下去就形成了⼀个层次化命名空间 ZNode 树，我们称为 ZNode Tree，它采⽤了类似⽂件系统的层级树状结构进⾏管理。⻅下图示例：

在 Zookeeper 中，每⼀个数据节点都是⼀个 ZNode，上图根⽬录下有两个节点，分别是：app1 和app2，其中 app1 下⾯⼜有三个⼦节点,所有ZNode按层次化进⾏组织，形成这么⼀颗树，ZNode的节点路径标识⽅式和Unix⽂件系统路径⾮常相似，都是由⼀系列使⽤斜杠（/）进⾏分割的路径表示，开发⼈员可以向这个节点写⼊数据，也可以在这个节点下⾯创建⼦节点。

ZNode的类型

刚刚已经了解到，Zookeeper的znode tree是由⼀系列数据节点组成的，那接下来，我们就对数据节点做详细讲解

Zookeeper 节点类型可以分为三⼤类：

• 持久性节点（Persistent）

• 临时性节点（Ephemeral）

• 顺序性节点（Sequential）

在开发中在创建节点的时候通过组合可以⽣成以下四种节点类型：持久节点、持久顺序节点、临时节点、临时顺序节点。不同类型的节点则会有不同的⽣命周期

**持久节点：**是Zookeeper中最常⻅的⼀种节点类型，所谓持久节点，就是指节点被创建后会⼀直存在服务器，直到删除操作主动清除

**持久顺序节点：**就是有顺序的持久节点，节点特性和持久节点是⼀样的，只是额外特性表现在顺序上。顺序特性实质是在创建节点的时候，会在节点名后⾯加上⼀个数字后缀，来表示其顺序。

**临时节点：**就是会被⾃动清理掉的节点，它的⽣命周期和客户端会话绑在⼀起，客户端会话结束，节点会被删除掉。与持久性节点不同的是，临时节点不能创建⼦节点。

**临时顺序节点：**就是有顺序的临时节点，和持久顺序节点相同，在其创建的时候会在名字后⾯加上数字后缀。

事务ID

⾸先，先了解，事务是对物理和抽象的应⽤状态上的操作集合。往往在现在的概念中，ሀ义上的事务通常指的是数据库事务，⼀般包含了⼀系列对数据库有序的读写操作，这些数据库事务具有所谓的ACID特性，即原⼦性（Atomic）、⼀致性（Consistency）、隔离性（Isolation）和持久性（Durability）。

⽽在ZooKeeper中，事务是指能够改变ZooKeeper服务器状态的操作，我们也称之为事务操作或更新操作，⼀般包括数据节点创建与删除、数据节点内容更新等操作。对于每⼀个事务请求，ZooKeeper都会为其分配⼀个全局唯⼀的事务ID，⽤ ZXID 来表示，通常是⼀个 64 位的数字。每⼀个 ZXID 对应⼀次更新操作，从这些ZXID中可以间接地识别出ZooKeeper处理这些更新操作请求的全局顺序

ZNode的状态信息

整个 ZNode 节点内容包括两部分：节点数据内容和节点状态信息。图中quota 是数据内容，其他的属于状态信息。那么这些状态信息都有什么含义呢？

cZxid 就是 Create ZXID，表示节点被创建时的事务ID。
ctime 就是 Create Time，表示节点创建时间。
mZxid 就是 Modified ZXID，表示节点最后⼀次被修改时的事务ID。
mtime 就是 Modified Time，表示节点最后⼀次被修改的时间。
pZxid 表示该节点的⼦节点列表最后⼀次被修改时的事务 ID。只有⼦节点列表变更才会更新 pZxid，⼦节点内容变更不会更新。
cversion 表示⼦节点的版本号。
dataVersion 表示内容版本号。
aclVersion 标识acl版本
ephemeralOwner 表示创建该临时节点时的会话 sessionID，如果是持久性节点那么值为 0
dataLength 表示数据⻓度。
numChildren 表示直系⼦节点数。

Watcher--数据变更通知

Zookeeper使⽤Watcher机制实现分布式数据的发布/订阅功能

⼀个典型的发布/订阅模型系统定义了⼀种 ⼀对多的订阅关系，能够让多个订阅者同时监听某⼀个主题对象，当这个主题对象⾃身状态变化时，会通知所有订阅者，使它们能够做出相应的处理。

在 ZooKeeper 中，引⼊了 Watcher 机制来实现这种分布式的通知功能。ZooKeeper 允许客户端向服务端注册⼀个 Watcher 监听，当服务端的⼀些指定事件触发了这个 Watcher，那么就会向指定客户端发送⼀个事件通知来实现分布式的通知功能。

整个Watcher注册与通知过程如图所示。

Zookeeper的Watcher机制主要包括客户端线程、客户端WatcherManager、Zookeeper服务器三部分。

具体⼯作流程为：客户端在向Zookeeper服务器注册的同时，会将Watcher对象存储在客户端的WatcherManager当中。当Zookeeper服务器触发Watcher事件后，会向客户端发送通知，客户端线程从WatcherManager中取出对应的Watcher对象来执⾏回调逻辑。

ACL--保障数据的安全

Zookeeper作为⼀个分布式协调框架，其内部存储了分布式系统运⾏时状态的元数据，这些元数据会直接影响基于Zookeeper进⾏构造的分布式系统的运⾏状态，因此，如何保障系统中数据的安全，从⽽避免因误操作所带来的数据随意变更⽽导致的数据库异常⼗分重要，在Zookeeper中，提供了⼀套完善的ACL（Access Control List）权限控制机制来保障数据的安全。

我们可以从三个⽅⾯来理解ACL机制：权限模式（Scheme）、授权对象（ID）、权限（Permission），通常使⽤"scheme: id : permission"来标识⼀个有效的ACL信息。

权限模式：Scheme

权限模式⽤来确定权限验证过程中使⽤的检验策略，有如下四种模式：

1. IP

IP模式就是通过IP地址粒度来进⾏权限控制，如"ip:192.168.0.110"表示权限控制针对该IP地址，同时IP模式可以⽀持按照⽹段⽅式进⾏配置，如"ip:192.168.0.1/24"表示针对192.168.0.*这个⽹段进⾏权限控制。

2. Digest

Digest是最常⽤的权限控制模式，要更符合我们对权限控制的认识，其使⽤"username:password"形式的权限标识来进⾏权限配置，便于区分不同应⽤来进⾏权限控制。

当我们通过“username:password”形式配置了权限标识后，Zookeeper会先后对其进⾏SHA-1加密和BASE64编码。

3. World

World是⼀种最开放的权限控制模式，这种权限控制⽅式⼏乎没有任何作⽤，数据节点的访问权限对所有⽤户开放，即所有⽤户都可以在不进⾏任何权限校验的情况下操作ZooKeeper上的数据。

另外，World模式也可以看作是⼀种特殊的Digest模式，它只有⼀个权限标识，即“world：anyone”。

4. Super

Super模式，顾名思义就是超级⽤户的意思，也是⼀种特殊的Digest模式。在Super模式下，超级⽤户可以对任意ZooKeeper上的数据节点进⾏任何操作。

授权对象：ID

授权对象指的是权限赋予的⽤户或⼀个指定实体，例如 IP 地址或是机器等。在不同的权限模式下，授权对象是不同的，表中列出了各个权限模式和授权对象之间的对应关系。

权限

权限就是指那些通过权限检查后可以被允许执⾏的操作。在ZooKeeper中，所有对数据的操作权限分为以下五⼤类：

• CREATE（C）：数据节点的创建权限，允许授权对象在该数据节点下创建⼦节点。
• DELETE（D）：⼦节点的删除权限，允许授权对象删除该数据节点的⼦节点。
• READ（R）：数据节点的读取权限，允许授权对象访问该数据节点并读取其数据内容或⼦节点列表等。
• WRITE（W）：数据节点的更新权限，允许授权对象对该数据节点进⾏更新操作。
• ADMIN（A）：数据节点的管理权限，允许授权对象对该数据节点进⾏ ACL 相关的设置操作。

2 ZooKeeper命令⾏操作

现在已经搭建起了⼀个能够正常运⾏的zookeeper服务了，所以接下来，就是来借助客户端来对zookeeper的数据节点进⾏操作

⾸先，进⼊到zookeeper的bin⽬录之后

通过zkClient进⼊zookeeper客户端命令⾏

./zkcli.sh 连接本地的zookeeper服务器
./zkCli.sh -server ip:port 连接指定的服务器

连接成功之后，系统会输出Zookeeper的相关环境及配置信息等信息。输⼊help之后，屏幕会输出可⽤的Zookeeper命令，如下图所示

创建节点

使⽤create命令，可以创建⼀个Zookeeper节点， 如

create [-s][-e] path data acl

其中，-s或-e分别指定节点特性，顺序或临时节点，若不指定，则创建持久节点；acl⽤来进⾏权限控制。

①创建顺序节点

使⽤ create -s /zk-test 123 命令创建zk-test顺序节点

执⾏完后，就在根节点下创建了⼀个叫做/zk-test的节点，该节点内容就是123，同时可以看到创建的zk-test节点后⾯添加了⼀串数字以示区别

②创建临时节点

使⽤ create -e /zk-temp 123 命令创建zk-temp临时节

临时节点在客户端会话结束后，就会⾃动删除，下⾯使⽤quit命令退出客户端

再次使⽤客户端连接服务端，并使⽤ls / 命令查看根⽬录下的节点

可以看到根⽬录下已经不存在zk-temp临时节点了

③创建永久节点

使⽤ create /zk-permanent 123 命令创建zk-permanent永久节点

可以看到永久节点不同于顺序节点，不会⾃动在后⾯添加⼀串数字

读取节点

与读取相关的命令有ls 命令和get 命令

ls命令可以列出Zookeeper指定节点下的所有⼦节点，但只能查看指定节点下的第⼀级的所有⼦节点；

ls path

其中，path表示的是指定数据节点的节点路径

get命令可以获取Zookeeper指定节点的数据内容和属性信息。

get path

若获取根节点下⾯的所有⼦节点，使⽤ls / 命令即可

若想获取/zk-permanent的数据内容和属性，可使⽤如下命令：get /zk-permanent

从上⾯的输出信息中，我们可以看到，第⼀⾏是节点/zk-permanent 的数据内容，其他⼏⾏则是创建该节点的事务ID（cZxid）、最后⼀次更新该节点的事务ID（mZxid）和最后⼀次更新该节点的时间（mtime）等属性信息

更新节点

使⽤set命令，可以更新指定节点的数据内容，⽤法如下

set path data [version]

其中，data就是要更新的新内容，version表示数据版本，在zookeeper中，节点的数据是有版本概念的，这个参数⽤于指定本次更新操作是基于Znode的哪⼀个数据版本进⾏的，如将/zk-permanent节点的数据更新为456，可以使⽤如下命令：set /zk-permanent 456

现在dataVersion已经变为1了，表示进⾏了更新

删除节点

使⽤delete命令可以删除Zookeeper上的指定节点，⽤法如下

delete path [version]

其中version也是表示数据版本，使⽤delete /zk-permanent 命令即可删除/zk-permanent节点

可以看到，已经成功删除/zk-permanent节点。值得注意的是，若删除节点存在⼦节点，那么⽆法删除该节点，必须先删除⼦节点，再删除⽗节点

3 Zookeeper的api使⽤

Zookeeper作为⼀个分布式框架，主要⽤来解决分布式⼀致性问题，它提供了简单的分布式原语，并且对多种编程语⾔提供了API，所以接下来重点来看下Zookeeper的java客户端API使⽤⽅式

Zookeeper API共包含五个包，分别为：

（1）org.apache.zookeeper

（2）org.apache.zookeeper.data

（3）org.apache.zookeeper.server

（4）org.apache.zookeeper.server.quorum

（5）org.apache.zookeeper.server.upgrade

其中org.apache.zookeeper，包含Zookeeper类，他是我们编程时最常⽤的类⽂件。这个类是Zookeeper客户端的主要类⽂件。如果要使⽤Zookeeper服务，应⽤程序⾸先必须创建⼀个Zookeeper实例，这时就需要使⽤此类。⼀旦客户端和Zookeeper服务端建⽴起了连接，Zookeeper系统将会给本次连接会话分配⼀个ID值，并且客户端将会周期性的向服务器端发送⼼跳来维持会话连接。只要连接有效，客户端就可以使⽤Zookeeper API来做相应处理了。

准备⼯作：导⼊依赖

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

建⽴会话

public class CreateSession implements Watcher
{
    //countDownLatch这个类使⼀个线程等待,主要不让main⽅法结束
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    public static void main(String[] args) throws InterruptedException, IOException
        {
            /*

            客户端可以通过创建⼀个zk实例来连接zk服务器

            new Zookeeper(connectString,sesssionTimeOut,Wather)

            connectString: 连接地址：IP：端⼝

            sesssionTimeOut：会话超时时间：单位毫秒

            Wather：监听器(当特定事件触发监听时，zk会通过watcher通知到客户端)

            */
            ZooKeeper zooKeeper = new ZooKeeper("10.211.55.4:2181", 5000, new CreateSession());
            System.out.println(zooKeeper.getState());
            countDownLatch.await();
            //表示会话真正建⽴
            System.out.println("=========Client Connected to zookeeper === === === = ");
        }
    // 当前类实现了Watcher接⼝，重写了process⽅法，该⽅法负责处理来⾃Zookeeper服务端的watcher通知， 在收到服务端发送过来的SyncConnected事件之后， 解除主程序在CountDownLatch上的等待阻塞，⾄ 此， 会话创建完毕
    public void process(WatchedEvent watchedEvent)
    {
        //当连接创建了，服务端发送给客户端SyncConnected事件
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected)
        {
            countDownLatch.countDown();
        }
    }
}

注意，ZooKeeper 客户端和服务端会话的建⽴是⼀个异步的过程，也就是说在程序中，构造⽅法会在处理完客户端初始化⼯作后⽴即返回，在⼤多数情况下，此时并没有真正建⽴好⼀个可⽤的会话，在会话的⽣命周期中处于“CONNECTING”的状态。 当该会话真正创建完毕后ZooKeeper服务端会向会话对应的客户端发送⼀个事件通知，以告知客户端，客户端只有在获取这个通知之后，才算真正建⽴了会话。

创建节点

public class CreateNote implements Watcher
{
    //countDownLatch这个类使⼀个线程等待,主要不让main⽅法结束
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws Exception
    {
        zooKeeper = new ZooKeeper("10.211.55.4:2181", 5000, new CreateNote());
        countDownLatch.await();
    }
    public void process(WatchedEvent watchedEvent)
    {
        //当连接创建了，服务端发送给客户端SyncConnected事件
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected)
        {
            countDownLatch.countDown();
        }
        //调⽤创建节点⽅法
        try
        {
            createNodeSync();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
    private void createNodeSync() throws Exception
    {
        /**
        * path ：节点创建的路径
        * data[] ：节点创建要保存的数据，是个byte类型的
        * acl ：节点创建的权限信息(4种类型)
        * ANYONE_ID_UNSAFE : 表示任何⼈
        * AUTH_IDS ：此ID仅可⽤于设置ACL。它将被客户机验证的ID替换。
        * OPEN_ACL_UNSAFE ：这是⼀个完全开放的ACL(常⽤)-->world:anyone
        * CREATOR_ALL_ACL ：此ACL授予创建者身份验证ID的所有权限
        * createMode ：创建节点的类型(4种类型)
        * PERSISTENT：持久节点
        * PERSISTENT_SEQUENTIAL：持久顺序节点
        * EPHEMERAL：临时节点
        * EPHEMERAL_SEQUENTIAL：临时顺序节点
        String node = zookeeper.create(path,data,acl,createMode);
        */
        String node_PERSISTENT = zooKeeper.create("/lg_persistent", "持久节点内容 ".getBytes("utf-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        String node_PERSISTENT_SEQUENTIAL = zooKeeper.create("/lg_persistent_sequential", "持久节点内容".getBytes("utf-8"), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
        String node_EPERSISTENT = zooKeeper.create("/lg_ephemeral", "临时节点内容 ".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("创建的持久节点是:" + node_PERSISTENT);
        System.out.println("创建的持久顺序节点是:" + node_PERSISTENT_SEQUENTIAL);
        System.out.println("创建的临时节点是:" + node_EPERSISTENT);
    }
}

获取节点数据

public class GetNoteData implements Watcher
{
    //countDownLatch这个类使⼀个线程等待,主要不让main⽅法结束
    private static CountDownLatch countDownLatch = new CountDownLatch(1);
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws Exception
    {
        zooKeeper = new ZooKeeper("10.211.55.4:2181", 10000, new GetNoteDate());
        Thread.sleep(Integer.MAX_VALUE);
    }
    public void process(WatchedEvent watchedEvent)
    {
        //⼦节点列表发⽣变化时，服务器会发出NodeChildrenChanged通知，但不会把变化情况告诉给客户端
        // 需要客户端⾃⾏获取，且通知是⼀次性的，需反复注册监听
        if(watchedEvent.getType() == Event.EventType.NodeChildrenChanged)
        {
            //再次获取节点数据
            try
            {
                List < String > children = zooKeeper.getChildren(watchedEvent.getPath(), true);
                System.out.println(children);
            }
            catch (KeeperException e)
            {
                e.printStackTrace();
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
        //当连接创建了，服务端发送给客户端SyncConnected事件
        if(watchedEvent.getState() == Event.KeeperState.SyncConnected)
        {
            try
            {
                //调⽤获取单个节点数据⽅法
                getNoteDate();
                getChildrens();
            }
            catch (KeeperException e)
            {
                e.printStackTrace();
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }
    private static void getNoteData() throws Exception
    {
        /**
        * path : 获取数据的路径
        * watch : 是否开启监听
        * stat : 节点状态信息
        * null: 表示获取最新版本的数据
        * zk.getData(path, watch, stat);
        */
        byte[] data = zooKeeper.getData("/lg_persistent/lg-children", true, null);
        System.out.println(new String(data, "utf-8"));
    }
    private static void getChildrens() throws KeeperException,
        InterruptedException
        {
            /*

            path:路径

            watch:是否要启动监听，当⼦节点列表发⽣变化，会触发监听

            zooKeeper.getChildren(path, watch);

            */
            List < String > children = zooKeeper.getChildren("/lg_persistent", true);
            System.out.println(children);
        }
}

修改节点数据

public class updateNote implements Watcher
{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws Exception
    {
        zooKeeper = new ZooKeeper("10.211.55.4:2181", 5000, new updateNote());
        Thread.sleep(Integer.MAX_VALUE);
    }
    public void process(WatchedEvent watchedEvent)
    {
        //当连接创建了，服务端发送给客户端SyncConnected事件
        try
        {
            updateNodeSync();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
    private void updateNodeSync() throws Exception
    {
        /*

        path:路径

        data:要修改的内容 byte[]

        version:为-1，表示对最新版本的数据进⾏修改

        zooKeeper.setData(path, data,version);

        */
        byte[] data = zooKeeper.getData("/lg_persistent", false, null);
        System.out.println("修改前的值:" + new String(data));
        //修改 stat:状态信息对象 -1:最新版本
        Stat stat = zooKeeper.setData("/lg_persistent", "客户端修改内
            容 ".getBytes(), -1);
            byte[] data2 = zooKeeper.getData("/lg_persistent", false, null); System.out.println("修改后的值:" + new String(data2));
        }
    }

删除节点

public class DeleteNote implements Watcher
{
    private static ZooKeeper zooKeeper;
    public static void main(String[] args) throws Exception
    {
        zooKeeper = new ZooKeeper("10.211.55.4:2181", 5000, new DeleteNote());
        Thread.sleep(Integer.MAX_VALUE);
    }
    public void process(WatchedEvent watchedEvent)
    {
        //当连接创建了，服务端发送给客户端SyncConnected事件
        try
        {
            deleteNodeSync();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
    private void deleteNodeSync() throws KeeperException, InterruptedException
    {
        /*

        zooKeeper.exists(path,watch) :判断节点是否存在

        zookeeper.delete(path,version) : 删除节点

        */
        Stat exists = zooKeeper.exists("/lg_persistent/lg-children", false);
        System.out.println(exists == null ? "该节点不存在" : "该节点存在");
        zooKeeper.delete("/lg_persistent/lg-children", -1);
        Stat exists2 = zooKeeper.exists("/lg_persistent/lg-children", false);
        System.out.println(exists2 == null ? "该节点不存在" : "该节点存在");
    }
}

4 Zookeeper-开源客户端

ZkClient

ZkClient是Github上⼀个开源的zookeeper客户端，在Zookeeper原⽣API接⼝之上进⾏了包装，是⼀个更易⽤的Zookeeper客户端，同时，zkClient在内部还实现了诸如Session超时重连、Watcher反复注册等功能

接下来，还是从创建会话、创建节点、读取数据、更新数据、删除节点等⽅⾯来介绍如何使⽤zkClient这个zookeeper客户端

添加依赖：

在pom.xml⽂件中添加如下内容

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.2</version>
</dependency>

创建会话：

使⽤ZkClient可以轻松的创建会话，连接到服务端。

package com.hust.grid.leesf.zkclient.examples;
import java.io.IOException;
import org.I0Itec.zkclient.ZkClient;
public class CreateSession
{
    /*

    创建⼀个zkClient实例来进⾏连接

    注意：zkClient通过对zookeeperAPI内部包装，将这个异步的会话创建过程同步化了

    */
    public static void main(String[] args)
    {
        ZkClient zkClient = new ZkClient("127.0.0.1:2181");
        System.out.println("ZooKeeper session established.");
    }
}

运⾏结果：ZooKeeper session established.

结果表明已经成功创建会话。

创建节点

ZkClient提供了递归创建节点的接⼝，即其帮助开发者先完成⽗节点的创建，再创建⼦节点

package com.hust.grid.leesf.zkclient.examples;
import org.I0Itec.zkclient.ZkClient;
public class Create_Node_Sample
{
    public static void main(String[] args)
    {
        ZkClient zkClient = new ZkClient("127.0.0.1:2181");
        System.out.println("ZooKeeper session established.");
        //createParents的值设置为true，可以递归创建节点
        zkClient.createPersistent("/lg-zkClient/lg-c1", true);
        System.out.println("success create znode.");
    }
}

运⾏结果：success create znode.

结果表明已经成功创建了节点，值得注意的是，在原⽣态接⼝中是⽆法创建成功的（⽗节点不存在），但是通过ZkClient通过设置createParents参数为true可以递归的先创建⽗节点，再创建⼦节点

删除节点

ZkClient提供了递归删除节点的接⼝，即其帮助开发者先删除所有⼦节点（存在），再删除⽗节点。

package com.hust.grid.leesf.zkclient.examples;
import org.I0Itec.zkclient.ZkClient;
public class Del_Data_Sample
{
    public static void main(String[] args) throws Exception
    {
        String path = "/lg-zkClient/lg-c1";
        ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);
        zkClient.deleteRecursive(path);
        System.out.println("success delete znode.");
    }
}

运⾏结果: success delete znode.

结果表明ZkClient可直接删除带⼦节点的⽗节点，因为其底层先删除其所有⼦节点，然后再删除⽗节点

获取⼦节点

package com.hust.grid.leesf.zkclient.examples;
import java.util.List;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
public class Get_Children_Sample
{
    public static void main(String[] args) throws Exception
    {
        ZkClient zkClient = new ZkClient("127.0.0.1:2181", 5000);
        List < String > children = zkClient.getChildren("/lg-zkClient");
        System.out.println(children);
        //注册监听事件
        zkClient.subscribeChildChanges(path, new IZkChildListener()
            {
                public void handleChildChange(String parentPath, List < String > currentChilds) throws Exception
                {
                    System.out.println(parentPath + " 's child changed,
                        currentChilds: " + currentChilds);
                    }
                }); zkClient.createPersistent("/lg-zkClient"); Thread.sleep(1000); zkClient.createPersistent("/lg-zkClient/c1"); Thread.sleep(1000); zkClient.delete("/lg-zkClient/c1"); Thread.sleep(1000); zkClient.delete(path); Thread.sleep(Integer.MAX_VALUE);
        }
    }

运⾏结果：

/zk-book 's child changed, currentChilds:[]
/zk-book 's child changed, currentChilds:[c1]
/zk-book 's child changed, currentChilds:[]
/zk-book 's child changed, currentChilds:null

结果表明：

客户端可以对⼀个不存在的节点进⾏⼦节点变更的监听。

⼀旦客户端对⼀个节点注册了⼦节点列表变更监听之后，那么当该节点的⼦节点列表发⽣变更时，服务端都会通知客户端，并将最新的⼦节点列表发送给客户端

该节点本身的创建或删除也会通知到客户端。

获取数据（节点是否存在、更新、删除）

public class Get_Data_Sample
{
    public static void main(String[] args) throws InterruptedException
    {
        String path = "/lg-zkClient-Ep";
        ZkClient zkClient = new ZkClient("127.0.0.1:2181");
        //判断节点是否存在
        boolean exists = zkClient.exists(path);
        if(!exists)
        {
            zkClient.createEphemeral(path, "123");
        }
        //注册监听
        zkClient.subscribeDataChanges(path, new IZkDataListener()
        {
            public void handleDataChange(String path, Object data) throws
            Exception
            {
                System.out.println(path + "该节点内容被更新，更新后的内容" + data);
            }
            public void handleDataDeleted(String s) throws Exception
            {
                System.out.println(s + " 该节点被删除");
            }
        });
        //获取节点内容
        Object o = zkClient.readData(path);
        System.out.println(o);
        //更新
        zkClient.writeData(path, "4567");
        Thread.sleep(1000);
        //删除
        zkClient.delete(path);
        Thread.sleep(1000);
    }
}

运⾏结果：

123
/lg-zkClient-Ep该节点内容被更新，更新后的内容4567
/lg-zkClient-Ep 该节点被删除

结果表明可以成功监听节点数据变化或删除事件。

Curator客户端

curator是Netflix公司开源的⼀套Zookeeper客户端框架，和ZKClient⼀样，Curator解决了很多Zookeeper客户端⾮常底层的细节开发⼯作，包括连接重连，反复注册Watcher和NodeExistsException异常等，是最流⾏的Zookeeper客户端之⼀。从编码⻛格上来讲，它提供了基于Fluent的编程⻛格⽀持

添加依赖

在pom.xml⽂件中添加如下内容：

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.12.0</version>
</dependency>

创建会话

Curator的创建会话⽅式与原⽣的API和ZkClient的创建⽅式区别很⼤。Curator创建客户端是通过

CuratorFrameworkFactory⼯⼚类来实现的。具体如下：

1.使⽤CuratorFramework这个⼯⼚类的两个静态⽅法来创建⼀个客户端

public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)

public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)

其中参数RetryPolicy提供重试策略的接⼝，可以让⽤户实现⾃定义的重试策略，默认提供了以下实现，分别为ExponentialBackoffRetry（基于backoff的重连策略）、RetryNTimes（重连N次策略）、RetryForever（永远重试策略）、

2.通过调⽤CuratorFramework中的start()⽅法来启动会话

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",retryPolicy);
client.start();

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);
CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 5000,1000,retryPolicy);
client.start();

其实进⼀步查看源代码可以得知，其实这两种⽅法内部实现⼀样，只是对外包装成不同的⽅法。它们的底层都是通过第三个⽅法builder来实现的

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
private static CuratorFramework Client = CuratorFrameworkFactory
  .builder()
  .connectString("server1:2181,server2:2181,server3:2181")
  .sessionTimeoutMs(50000)
  .connectionTimeoutMs(30000)
  .retryPolicy(retryPolicy)
  .build();
client.start();

参数：

• connectString：zk的server地址，多个server之间使⽤英⽂逗号分隔开

• connectionTimeoutMs：连接超时时间，如上是30s，默认是15s

• sessionTimeoutMs：会话超时时间，如上是50s，默认是60s

• retryPolicy：失败重试策略

  • ExponentialBackoffRetry：构造器含有三个参数ExponentialBackoffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs)

    • baseSleepTimeMs：初始的sleep时间，⽤于计算之后的每次重试的sleep时间，

      • 计算公式：当前sleep时间=baseSleepTimeMs*Math.max(1, random.nextInt(1<<(retryCount+1)))

    • maxRetries：最⼤重试次数

    • maxSleepMs：最⼤sleep时间，如果上述的当前sleep计算出来⽐这个⼤，那么sleep⽤这个时间，默认的最⼤时间是Integer.MAX_VALUE毫秒。

  • 其他，查看org.apache.curator.RetryPolicy接⼝的实现类

  • start()：完成会话的创建

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class Create_Session_Sample
{
    public static void main(String[] args) throws Exception
    {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 5000, 3000, retryPolicy);
        client.start();
        System.out.println("Zookeeper session1 established. ");
        CuratorFramework client1 = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") //server地址
            .sessionTimeoutMs(5000) // 会话超时时间
            .connectionTimeoutMs(3000) // 连接超时时间
            .retryPolicy(retryPolicy) // 重试策略
            .namespace("base") // ᇿ⽴命名空间/base
            .build(); //
        client1.start();
        System.out.println("Zookeeper session2 established. ");
    }
}

运⾏结果：Zookeeper session1 established. Zookeeper session2 established

需要注意的是session2会话含有隔离命名空间，即客户端对Zookeeper上数据节点的任何操作都是相对/base⽬录进⾏的，这有利于实现不同的Zookeeper的业务之间的隔离

创建节点

curator提供了⼀系列Fluent⻛格的接⼝，通过使⽤Fluent编程⻛格的接⼝，开发⼈员可以进⾏⾃由组合来完成各种类型节点的创建。

下⾯简单介绍⼀下常⽤的⼏个节点创建场景。

（1）创建⼀个初始内容为空的节点

client.create().forPath(path);

Curator默认创建的是持久节点，内容为空。

（2）创建⼀个包含内容的节点

client.create().forPath(path,"我是内容".getBytes());

Curator和ZkClient不同的是依旧采⽤Zookeeper原⽣API的⻛格，内容使⽤byte[]作为⽅法参数。

（3）递归创建⽗节点,并选择节点类型

client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);

creatingParentsIfNeeded这个接⼝⾮常有⽤，在使⽤ZooKeeper 的过程中，开发⼈员经常会碰到NoNodeException 异常，其中⼀个可能的原因就是试图对⼀个不存在的⽗节点创建⼦节点。因此，开发⼈员不得不在每次创建节点之前，都判断⼀下该⽗节点是否存在——这个处理通常⽐较麻烦。在使⽤Curator 之后，通过调⽤creatingParentsIfNeeded 接⼝，Curator 就能够⾃动地递归创建所有需要的⽗节点。

下⾯通过⼀个实际例⼦来演示如何在代码中使⽤这些API。

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
public class Sample
{
    public static void main(String[] args) throws Exception
    {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") //server地址
            .sessionTimeoutMs(5000) // 会话超时时间
            .connectionTimeoutMs(3000) // 连接超时时间
            .retryPolicy(new ExponentialBackoffRetry(1000, 5)) //重试策略
            .build(); //
        client.start();
        System.out.println("Zookeeper session established. ");
        //添加节点
        String path = "/lg-curator/c1";
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        Thread.sleep(1000);
        System.out.println("success create znode" + path);
    }
}

运⾏结果：Zookeeper session established. success create znode/lg-curator/c1

其中，也创建了lg-curator/c1的⽗节点lg-curator节点。

删除节点

删除节点的⽅法也是基于Fluent⽅式来进⾏操作，不同类型的操作调⽤ 新增不同的⽅法调⽤即可。

（1）删除⼀个⼦节点

client.delete().forPath(path);

（2）删除节点并递归删除其⼦节点

client.delete().deletingChildrenIfNeeded().forPath(path);

（3）指定版本进⾏删除

client.delete().withVersion(1).forPath(path);

如果此版本已经不存在，则删除异常，异常信息如下。

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for

（4）强制保证删除⼀个节点

client.delete().guaranteed().forPath(path);

只要客户端会话有效，那么Curator会在后台持续进⾏删除操作，直到节点删除成功。⽐如遇到⼀些⽹络异常的情况，此guaranteed的强制删除就会很有效果。

演示实例：

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class Sample
{
    public static void main(String[] args) throws Exception
    {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") //server地址
            .sessionTimeoutMs(5000) // 会话超时时间
            .connectionTimeoutMs(3000) // 连接超时时间
            .retryPolicy(new ExponentialBackoffRetry(1000, 5)) //重试策略
            .build(); //
        client.start();
        System.out.println("Zookeeper session established. ");
        //删除节点
        String path = "/lg-curator";
        client.delete().deletingChildrenIfNeeded().withVersion(-1).forPath(path);
        System.out.println("success create znode" + path);
    }
}

运⾏结果：Zookeeper session established. success create znode/lg-curator

结果表明成功删除/lg-curator节点

获取数据

获取节点数据内容API相当简单，同时Curator提供了传⼊⼀个Stat变量的⽅式来存储服务器端返回的最新的节点状态信息

// 普通查询
client.getData().forPath(path);
// 包含状态查询
Stat stat = new Stat();
client.getData().storingStatIn(stat).forPath(path);
演示：
package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class Get_Node_Sample
{
    public static void main(String[] args) throws Exception
    {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") //server地址
            .sessionTimeoutMs(5000) // 会话超时时间
            .connectionTimeoutMs(3000) // 连接超时时间
            .retryPolicy(new ExponentialBackoffRetry(1000, 5)) //重试策略
            .build(); //
        client.start();
        System.out.println("Zookeeper session established. ");
        //添加节点
        String path = "/lg-curator/c1";
        client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path, "init".getBytes());
        System.out.println("success create znode" + path);
        //获取节点数据
        Stat stat = new Stat();
        byte[] bytes = client.getData().storingStatIn(stat).forPath(path);
        System.out.println(new String(bytes));
    }
}

运⾏结果：Zookeeper session established. success create znode/lg-curator/c1 init

结果表明成功获取了节点的数据

更新数据

更新数据，如果未传⼊version参数，那么更新当前最新版本，如果传⼊version则更新指定version，如果version已经变更，则抛出异常。

// 普通更新
client.setData().forPath(path,"新内容".getBytes());

// 指定版本更新
client.setData().withVersion(1).forPath(path);

版本不⼀致异常信息：

org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for

案例演示：

package com.hust.grid.leesf.curator.examples;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class Set_Node_Sample
{
    public static void main(String[] args) throws Exception
    {
        CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181") //server地址
            .sessionTimeoutMs(5000) // 会话超时时间
            .connectionTimeoutMs(3000) // 连接超时时间
            .retryPolicy(new ExponentialBackoffRetry(1000, 5)) //重试策略
            .build(); //
        client.start();
        System.out.println("Zookeeper session established. ");
        String path = "/lg-curator/c1";
        //获取节点数据
        Stat stat = new Stat();
        byte[] bytes = client.getData().storingStatIn(stat).forPath(path);
        System.out.println(new String(bytes));
        //更新节点数据
        int version = client.setData().withVersion(stat.getVersion()).forPath(path).getVersion();
        System.out.println("Success set node for : " + path + ", new version: " + version);
        client.setData().withVersion(stat.getVersion()).forPath(path).getVersion();
    }
}

运⾏结果：

Zookeeper session established.
init
Success set node for : /lg-curator/c1, new version: 1
Exception in thread "main"
org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode =
BadVersion for /lg-curator/c1

结果表明当携带数据版本不⼀致时，⽆法完成更新操作。