Zookeeper集群及基础API_zookeeper api

目录

 

1、Zookeeper简介

2、Zookeeper概念

2.1 重要概念

2.2 Session会话（心跳检测）

2.3 ZNode数据节点

2.4 ZNode数据版本

2.5 Watcher-事件监听器

2.6 ACL策略

3、Zookeeper集群安装配置

3.1 准备工作

3.2 下载解压

3.3 修改配置

3.4 启动测试

4、Zookeeper客户端操作命令

5、Zookeeper API操作

5.1 引入POM依赖坐标

5.2 创建Seesion会话

5.3 创建ZNode节点

5.4 查找ZNode节点

5.5 删除Znode节点

5.6 操作监听

5.7 异步回调

6、Curator API操作

6.1 POM依赖坐标

6.2 创建Curator客户端

6.3 创建ZNode节点

6.4 查询ZNode节点

6.5 Transaction事务

6.6 异步接口回调

6.7 分布式计数器

6.8 cache操作监听

---

1、Zookeeper简介

Zookeeper是一个典型的分布式数据一致性解决方案，分布式应用程序可以基于Zookeeper实现诸如数据发布/订阅，负载均衡，命名服务，分布式协调/通知，集群管理，Master选举，分布式锁和分布式队列等功能。

Zookeeper一个最常用的使用场景就是用于担任生产者和服务消费者的注册中心，提供发布/订阅服务。服务生产者将自己提供的服务注册到Zookeeper中心，服务消费者在进行服务调用的时候先到Zookeeper中查找服务，获取到服务生产者的详细信息之后，再去调用服务生产者的内容与数据，如在Dubbo中，Zookeeper就担任注册中心的角色。

在使用Zookeeper的时候，最好使用集群版Zookeeper，而且最好使用集群版服务器构成Zookeeper集群。Zookeeper容错中，当宕调几个Zookeeper服务器后，剩下的个数必须大于宕掉个数，整个Zookeeper才能依然使用。假如集群中有n台Zookeeper服务器，那么也就是剩下的服务器数必须大于n/2，所以当n是奇数的时候则宕掉的最多为(n-1)/2,当n为偶数时，宕掉的最多为(n-2)/2，而最多的宕机数量都是一样的。比如假如我们有3台，那么最大允许宕掉1台zookeeper服务器，如果我们有4台的的时候也同样只允许宕掉1台。假如我们有5台，那么最大允许宕掉2台zookeeper服务器，如果我们有6台的的时候也同样只允许宕掉2台。所以使用奇数台zookeeper就够了。

为什么使用奇数，偶数的话浪费了一台机器的钱啊！

2、Zookeeper概念

2.1 重要概念

（1）Zookeeper本身就是一个分布式程序，只要半数以上节点存活，Zookeeper就能正常提供服务（过半原则）。

（2）Zookeeper将数据保存在内存中，保证了高吞吐和低延迟，但是内存限制了能够存储的容量，这也限制了Znode中存储的数据量较小的原因。

（3）Zookeeper是高性能的，在读多于写的应用程序中才能发挥最高性能，因为写会导致所有服务器进入同步状态。

（4）Zookeeper有临时节点的概念，当创建临时节点的客户端会话一直保持活动，临时节点就一直存在，而当会话终结时，临时节点被删除。持久节点是指一旦这个ZNode被创建，除非主动进行ZNode的移除操作，否则这个ZNode将一直保存在Zookeeper上。

（5）Zookeeper底层其实提供了两个功能：

管理用户程序提交的数据；

为用户程序提供节点监听服务。

2.2 Session会话（心跳检测）

Session是指Zookeeper服务器与客户端的会话。在Zookeeper中，一个客户端连接是指客户端和服务器之间的一个TCP长连接。客户端启动的时候，首先会与服务器建立一个TCP连接，从第一次连接建立开始，客户端会话的生命周期就开始了。通过这个连接，客户端能够通过心跳检测与服务器保持有效的会话，也能够向Zookeeper服务器发送请求并接受响应，同时还能够通过该连接接收来自服务器的Watch事件通知。

Session的Session Timeout可以设置客户端会话的超时事件。由于服务压力大，网络故障或者客户端主动断开连接等各种原因导致客户端连接断开时，只要在Session Timeout规定的时间内能够重新连接上集群中任意一台服务器，那么之前创建的会话仍然有效。

在为客户端创建会话之前，服务端首先会为每个客户端都分配一个Session ID,由于Session ID是Zookeeper会话的一个重要标识，许多与会话相关的运行机制都是基于Session ID的，因此，无论是哪台服务器为客户端分配的Session ID,都需要保证全局唯一。

2.3 ZNode数据节点

在Zookeeper中，节点分为两类，第一类是指构成集群的机器，称为机器节点，第二类是指数据模型中的数据单元，称为数据节点-ZNode。

Zookeeper将所有数据存储在内存中，数据模型是一棵树，由"/"进行分割的路径，就是一个ZNode，每个节点上都会保存自己的数据内容，同时还会保存一系列属性信息。

Zookeeper中ZNode分为两类，分为持久节点和临时节点。同时Zookeepe还允许为每个节点添加一个属性：SEQUENTIAL。一旦节点被标记上这个属性，那么这个节点被创建的时候，Zookeeper会自动在其节点名后追加上一个整型数字，这个整型数字是由父节点维护的自增数字。

2.4 ZNode数据版本

Zookeeper的每个ZNode上都会存储数据，对应于每个ZNode，Zookeeper都会为其维护一个State数据结构，State中记录了这个ZNode的三个数据版本，分别是Version-当前ZNode的版本，cversion-当前ZNode子节点的版本和aversion-当前ZNode的ACL版本。

2.5 Watcher-事件监听器

Watcher-事件监听器，是Zookeeper中一个很重要的特性。Zookeeper允许用于在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，Zookeeper服务端会将事件通知到感兴趣的客户端上去，该机制是Zookeeper实现分布式协调服务的重要特性。

2.6 ACL策略

Zookeeper采用ACL策略来进行权限控制，分为：

（1）CREATE：创建子节点的权限。

（2）READ：获取节点数据和子节点列表的权限。

（3）WRITE：更新节点数据的权限。

（4）DELETE：删除子节点的权限。

（5）ADMIN：设置节点的ACL权限。

其中CREATE和DELETE这两种权限都是针对子节点的控制权限。

3、Zookeeper集群安装配置

集群高可用架构图：

图中每一个Server代表一个安装Zookeeper服务的服务器。组成 ZooKeeper 服务的服务器都会在内存中维护当前的服务器状态，并且每台服务器之间都互相保持着通信。集群间通过 Zab 协议（Zookeeper Atomic Broadcast）来保持数据的一致性。

Zookeeper 集群中的角色

• Leader 领导者 ：Leader 节点负责Zookeeper集群内部投票的发起和决议（一次事务操作），更新系统的状态；同时它也能接收并且响应Client端发送的请求。Zookeeper中，能改变ZooKeeper服务器状态的操作称为事务操作。一般包括数据节点创建与删除、数据内容更新和客户端会话创建与失效等操作。**

• Learner 学习者

  • Follower 跟随者 ： Follower 节点用于接收并且响应Client端的请求，如果是事务操作，会将请求转发给Leader节点，发起投票，参与集群的内部投票，

  • Observer 观察者：Observer 节点功能和Follower相同，只是Observer 节点不参与投票过程，只会同步Leader节点的状态。

• Client 客户端

Zookeeper 通过复制来实现高可用。在上一章提到的集群模式（replicated mode）下，以Leader节点为准，Zookeeper的ZNode树上面的每一个修改都会被同步（复制）到其他的Server 节点上面。

3.1 准备工作

准备三台虚拟机：

192.168.223.128，192.168.223.129，192.168.223.130

3.2 下载解压

官网地址：https://zookeeper.apache.org/releases.html

#三台虚拟机分别执行如下命令安装相同版本的zookeeper
wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.6.1/apache-zookeeper-3.4.6-bin.tar.gz #下载
tar -zxvf apache-zookeeper-3.4.6-bin.tar.gz #解压
mv apache-zookeeper-3.4.6-bin zookeeper-3.4.6 #改个名字好看些

3.3 修改配置

在三台虚拟机zookeeper目录下分别将/conf/zoo_sample.cfg文件名称改为zoo.cfg

cd zookeeper-3.4.6/
mv conf/zoo_sample.cfg conf/zoo.cfg
​
#在zookeeper-3.4.6目录下分别创建一个data和logs目录
mkdir data
mkdir logs
​
#修改zoo.cfg文件配置
vim conf/zoo.cfg
---------------------------------#配置如下#---------------------------------------------
​
tickTime=2000 #作为 Zookeeper #服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime #时间就会发送一个心跳。
initLimit=10 #集群中的follower服务器(F)与leader服务器(L)之间初始连接时能容忍的最多心跳数（tickTime的数量）
syncLimit=5 #集群中的follower服务器与leader服务器之间请求和应答（数据同步）之间能容忍的最多心跳数（tick#Time的数量）。
​
dataDir=/usr/local/zookeeper-3.4.6/data #数据持久化目录
dataLogDir=/usr/local/zookeeper-3.4.6/logs #日志目录
clientPort=2181 #客户端连接 Zookeeper 服务器的端口 用默认就行
server.1=192.168.223.128:4000:5000
server.2=192.168.223.129:4000:5000
server.3=192.168.223.130:4000:5000
​
#创建多个节点集群时，在dataDir目录下必须创建myid文件，myid文件用于zookeeper验证server序号等，myid文件只有一行，并且为当前server的序号，例如server.1的myid就是1，server.2的myid就是#2等。
#server.A=B:C:D;其中 A 是一个数字，表示这个是第几号服务器；B 是这个服务器的 ip 地址；C #表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；D 表示的是万一集群中的 #Leader 服务器挂了，需要一个端口来重新进行选举，选出一个新的 #Leader，而这个端口就是用来执行选举时服务器相互通信的端口。如果是伪集群的配置方式，由于 #B 都是一样，所以不同的 Zookeeper 实例通信端口号不能一样，所以要给它们分配不同的端口号
------------------------------------------------------------------------------------------
#增加服务器号myid文件到data目录
vim data/myid
#192.168.223.128服务器写入：1
#192.168.223.129服务器写入：2
#192.168.223.130服务器写入：3

3.4 启动测试

#三台机器分别开启4000,5000和2181端口或者关闭防火墙
 #开启端口
firewall-cmd --zone=public --add-port=4000/tcp --permanent    
firewall-cmd --zone=public --add-port=5000/tcp --permanent
firewall-cmd --zone=public --add-port=2181/tcp --permanent
#重启防火墙
firewall-cmd --reload
​
#分别启动三个服务
./bin/zkServer.sh start
​
#查看zookeeper状态
./bin/zkServer.sh status

1、三台只开启128的时候，是不能正常提供服务的

2、继续开启129，130

 

 

3、将leader关闭，会重新选举新的leader

 

4、Zookeeper客户端操作命令

#启动客户端
./bin/zkCli.sh -server 192.168.223.128:2181
#PS：我们可以使用 help命令来查看帮助：
[zk: 192.168.223.128:2181(CONNECTED) 0] help
ZooKeeper -server host:port cmd args
        stat path [watch]
        set path data [version]
        ls path [watch]
        delquota [-n|-b] path
        ls2 path [watch]
        setAcl path acl
        setquota -n|-b val path
        history
        redo cmdno
        printwatches on|off
        delete path [version]
        sync path
        listquota path
        rmr path
        get path [watch]
        create [-s] [-e] path data acl
        addauth scheme auth
        quit
        getAcl path
        close
        connect host:port
​
#常用命令
​
#ls -- 查看某个目录包含的所有文件，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] ls /
​
#ls2 -- 查看某个目录包含的所有文件，与ls不同的是它查看到time、version等信息，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] ls2 /
​
#create -- 创建znode，并设置初始内容，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] create /test "test"
Created /test
​
#get -- 获取znode的数据，如下：
[zk: 127.0.0.1:2181(CONNECTED) 1] get /test
​
#set -- 修改znode内容，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] set /test "ricky"
​
#delete -- 删除znode，例如：
[zk: 127.0.0.1:2181(CONNECTED) 1] delete /test
​
#quit -- 退出客户端

 

5、Zookeeper API操作

5.1 引入POM依赖坐标

    <dependencies>
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.5.7</version>
        </dependency>
​
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

 

5.2 创建Seesion会话

 private static CountDownLatch countDownLatch = new CountDownLatch(1);
 private static ZooKeeper zooKeeper = null;
 private static final String hosts = "192.168.223.128:2181,192.168.223.129:2181,192.168.223.130:2181";
    /**
   * 创建会话（最基础的实例）
   *
   * @throws Exception Exception
   */
  public static void createSession1() throws Exception {
    //hosts 集群地址 ；5000 连接超时时间 ；new ZooKeeperApi() Watcher监听器
    zooKeeper = new ZooKeeper(hosts, 5000, new ZooKeeperWatcher());
    System.out.println("ZooKeeper.state：" + zooKeeper.getState());
    // 所有线程执行完毕
    countDownLatch.await();
    System.out.println("ZooKeeper session会话创建完成。");
  }
  
  /**
   * 创建会话（可复用sessionId的实例）
   *
   * @throws Exception Exception
   */
  public static void createSession2() throws Exception {
    createSession1();//完成基础会话实例
    long sessionId = zooKeeper.getSessionId();
    byte[] sessionPasswd = zooKeeper.getSessionPasswd();
    System.out.println(String.format("首次获取sessionId：%s，sessionPasswd：%s", sessionId, sessionPasswd));
    // 使用sessionId
    zooKeeper = new ZooKeeper(hosts, 5000, new ZooKeeperWatcher(), sessionId, sessionPasswd);
    System.out.println("ZooKeeper.state session：" + zooKeeper.getState());
    Thread.sleep(Integer.MAX_VALUE);
  }

5.3 创建ZNode节点

/**
   * 创建节点（同步）
   * ZooKeeper CreateMode节点类型说明：
   *  1.PERSISTENT：持久型
   *  2.PERSISTENT_SEQUENTIAL：持久顺序型 会在节点名后增加一串数字
   *  3.EPHEMERAL：临时型
   *  4.EPHEMERAL_SEQUENTIAL：临时顺序型 会在节点名后增加一串数字
   *    1、2种类型客户端断开后不会消失
   *    3、4种类型客户端断开后超时时间内没有新的连接节点将会消失
   * @throws Exception Exception
   */
  public static void createNodeSync() throws Exception {
    String path = "/test1";
    createSession1();
    String path1 = zooKeeper.create(path, "abc".getBytes()
            , ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    System.out.println("节点创建成功：" + path1);
    String path2 = zooKeeper.create(path, "def".getBytes()
            , ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
    System.out.println("节点创建成功：" + path2);
  }
​
​
  /**
   * 创建节点（异步）
   * 同步接口创建节点时需要考虑接口抛出异常的情况，
   * 异步接口的异常体现在回调函数的ResultCode响应码中，比同步接口更健壮。
   *
   * @throws Exception Exception
   */
  public static void createNodeASync() throws Exception {
    String path = "/test2";
    createSession1();
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL,
            new ZookeeperCallBack(), "ZooKeeper async create znode.");
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
            new ZookeeperCallBack(), "ZooKeeper async create znode.");
    Thread.sleep(Integer.MAX_VALUE);
  }

5.4 查找ZNode节点

/**
   * 获取子节点（同步）
   *
   * @throws Exception Exception
   */
  public static void getChildrenNodeSync() throws Exception {
    String path = "/zk-getChildren-sync-test";
    createSession1();
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    zooKeeper.create(path + "/children1", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    List<String> childrenList = zooKeeper.getChildren(path, true);
    System.out.println("获取子节点：" + childrenList);
    Thread.sleep(Integer.MAX_VALUE);
  }
​
  /**
   * 获取子节点（异步）
   *
   * @throws Exception Exception
   */
  public static void getChildrenNodeASync() throws Exception {
    String path = "/zk-getChildren-async-test";
    createSession1();
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    zooKeeper.create(path + "/children", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    zooKeeper.getChildren(path, true, new ChildrenCallBack(), "异步获取子节点");
    Thread.sleep(Integer.MAX_VALUE);
  }

 

5.5 删除Znode节点

/**
   * 删除节点（同步）
   * 注：只允许删除叶子节点，不能直接删除根节点
   *
   * @throws Exception Exception
   */
  public static void deleteNodeSync() throws Exception {
    String path = "/test1/zk-delete-znode-test1";
    createSession1();
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    zooKeeper.delete(path, -1);
    Thread.sleep(Integer.MAX_VALUE);
  }
​
  /**
   * 删除节点（异步）
   * 注：只允许删除叶子节点，不能直接删除根节点
   *
   * @throws Exception Exception
   */
  public void deleteNodeAsync() throws Exception {
    String path = "/test1/zk-delete-znode-test2";
    createSession1();
    zooKeeper.create(path, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    zooKeeper.delete(path, -1, new ZookeeperCallBack(), "ZooKeeper async delete znode");
    Thread.sleep(Integer.MAX_VALUE);
  }

5.6 操作监听

新建一个监听类，用来监听zookeeper的API操作

​
package com.ydt.zookeeper.watcher;
​
import com.ydt.zookeeper.api.ZooKeeperApi;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
​
/**
 * 功能描述：监听会话创建，节点创建、删除，数据读取、更新，权限控制等
 */
public class ZooKeeperWatcher implements Watcher {
    private static Stat stat = new Stat();
​
    /**
     * 接收事件通知
     *
     * @param event 事件通知
     */
    public void process(WatchedEvent event) {
        System.out.println("Receive WatchedEvent：" + event);
        try {
            if (Event.KeeperState.SyncConnected == event.getState()) {
                System.out.println("通知：会话连接成功");
                if (Event.EventType.None == event.getType() && null == event.getPath()) {
                    System.out.println("进入会话初始状态");
                    // 释放所有等待的线程,开始执行业务
                    ZooKeeperApi.countDownLatch.countDown();
                } else if (event.getType() == Event.EventType.NodeCreated) {
                    System.out.println("节点创建通知：" + event.getPath());
                } else if (event.getType() == Event.EventType.NodeDataChanged) {
                    System.out.println("节点的数据变更通知" + event.getPath());
                    System.out.println("czxid=" + stat.getCzxid()
                            + "，mzxid=" + stat.getMzxid() + "，version=" + stat.getVersion());
                } else if (event.getType() == Event.EventType.NodeChildrenChanged) {
                    System.out.println("子节点的数据变更通知：" + event.getPath());
                } else if (event.getType() == Event.EventType.NodeDeleted) {
                    System.out.println("节点删除通知：" + event.getPath());
                } else {
                    System.out.println("未知事件通知类型：" + event.getType());
                }
            } else if (Event.KeeperState.Disconnected == event.getState()) {
                System.out.println("通知：会话连接失败");
            } else if (Event.KeeperState.AuthFailed == event.getState()) {
                System.out.println("通知：会话认证失败");
            } else if (Event.KeeperState.Expired == event.getState()) {
                System.out.println("通知：会话过期");
            } else {
                System.out.println("未知的通知状态：" + event.getState());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
​
​
}

5.7 异步回调

主要用来跟踪异步操作结果并做相应的处理

package com.ydt.zookeeper.callback;
​
import org.apache.zookeeper.AsyncCallback;
​
/**
   * 创建异步回调
   */
public class ZookeeperCallBack implements AsyncCallback.StringCallback {
    /**
     * @param rc   服务端响应码 0：接口调用成功，-4：客户端与服务端连接已断开，-110：指定节点已存在，-112：会话已过期
     * @param path 调用接口时传入的节点路径（原样输出）
     * @param ctx  调用接口时传入的ctx值（原样输出）
     * @param name 实际在服务端创建的节点名
     */
    public void processResult(int rc, String path, Object ctx, String name) {
      System.out.println("创建结果：rc=" + rc + "，path=" + path + "，ctx=" + ctx + "，name=" + name);
      switch (rc) {
        case 0:
          System.out.println("节点创建成功：" + name);
          break;
        case -4:
          System.out.println("客户端与服务端连接已断开");
          break;
        case -110:
          System.out.println("指定节点已存在");
          break;
        case -112:
          System.out.println("会话已过期");
          break;
        default:
          System.out.println("服务端响应码" + rc + "未知");
          break;
      }
    }
  }

package com.ydt.zookeeper.callback;
​
import org.apache.zookeeper.AsyncCallback;
​
/**
 * 删除节点异步回调
 */
public class DeleteCallBack implements AsyncCallback.VoidCallback {
    /**
     * @param rc   服务端响应码 0：接口调用成功，-4：客户端与服务端连接已断开，-110：指定节点已存在，-112：会话已过期
     * @param path 调用接口时传入的节点路径（原样输出）
     * @param ctx  调用接口时传入的ctx值（原样输出）
     */
    public void processResult(int rc, String path, Object ctx) {
      System.out.println("删除结果：rc=" + rc + "，path=" + path + "，ctx=" + ctx);
      switch (rc) {
        case 0:
          System.out.println("节点删除成功");
          break;
        case -4:
          System.out.println("客户端与服务端连接已断开");
          break;
        case -112:
          System.out.println("会话已过期");
          break;
        default:
          System.out.println("服务端响应码" + rc + "未知");
          break;
      }
    }
  }

package com.ydt.zookeeper.callback;
​
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.data.Stat;
​
import java.util.List;
​
/**
   * 获取子节点异步回调
   */
public class ChildrenCallBack implements AsyncCallback.Children2Callback {
    /**
     * @param rc           服务端响应码 0：接口调用成功，-4：客户端与服务端连接已断开，-110：指定节点已存在，-112：会话已过期
     * @param path         调用接口时传入的节点路径（原样输出）
     * @param ctx          调用接口时传入的ctx值（原样输出）
     * @param childrenList 子节点列表
     * @param stat         节点状态，由服务器端响应的新stat替换
     */
    public void processResult(int rc, String path, Object ctx, List<String> childrenList, Stat stat) {
      System.out.println("获取结果：rc=" + rc + "，path=" + path + "，ctx=" + ctx + "，childrenList=" + childrenList + "，stat=" + stat);
      switch (rc) {
        case 0:
          System.out.println("子节点获取成功：" + childrenList);
          break;
        case -4:
          System.out.println("客户端与服务端连接已断开");
          break;
        case -112:
          System.out.println("会话已过期");
          break;
        default:
          System.out.println("服务端响应码" + rc + "未知");
          break;
      }
    }
  }

 

6、Curator API操作

Curator是Netflix公司开源的一套zookeeper客户端框架，解决了很多Zookeeper客户端非常底层的细节开发工作

Curator包含了几个包:

curator-framework：对zookeeper的底层api的一些封装

curator-client：提供一些客户端的操作，例如重试策略等

curator-recipes：封装了一些高级特性，如：Cache事件监听、选举、分布式锁、分布式计数器、分布式Barrier等

Maven依赖(使用curator的版本：2.12.0，对应Zookeeper的版本为：3.4.x，如果跨版本会有兼容性问题，很有可能导致节点操作失败)：

6.1 POM依赖坐标

<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13</version>
            <scope>test</scope>
        </dependency>
​
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>2.12.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
​
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>2.12.0</version>
        </dependency>
​
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>2.12.0</version>
        </dependency>
​
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>
​
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.25</version>
        </dependency>
​
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.25</version>
        </dependency>
    </dependencies>

6.2 创建Curator客户端

    CuratorFramework client = null;
​
    @Before
    public void init(){
        //核心参数变为流式设置
        client = CuratorFrameworkFactory
                .builder()
                .connectString("192.168.223.128:2181,192.168.223.129:2181,192.168.223.130:2181")
                /*
                    重连机制:
                        1、RetryOneTime N久后重连一次
                        2、RetryNTimes N久后重连N次
                        3、RetryForever N久后一直重连
                        4、RetryUntilElapsed 重连N次，与RetryOneTime区别是不能超过最长等待时间
                */
                .retryPolicy(new RetryOneTime(3000))
                .sessionTimeoutMs(5000)//超时时间
                .namespace("curator") //命名空间,相当于根节点root，默认为root
                .build();
        client.start();
        System.out.println(client.getState());
    }
​
    @After
    public void close(){
        client.close();
    }

6.3 创建ZNode节点

@Test
    public void createNode1() throws Exception {
        client.create()
                //节点类型
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath("/node1","node1".getBytes());
    }
​
    @Test
    public void createNode2() throws Exception {
        /**
         * 自定义权限列表
         */
        List<ACL> list = new ArrayList<ACL>();
        Id id = new Id("ip","192.168.223.130");
        list.add(new ACL(ZooDefs.Perms.DELETE,id));//给130节点赋予删除的权限（同步后所有节点都有这个权限了）
        client.create()
                //节点类型
                .withMode(CreateMode.PERSISTENT)
                .withACL(list)
                .forPath("/node2","node2".getBytes());
    }
​
    @Test
    public void createNode3() throws Exception {
        /**
         * 递归创建节点树
         */
        client.create()
                //如果需要创建父节点
                .creatingParentsIfNeeded()
                //节点类型
                .withMode(CreateMode.PERSISTENT)
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                .forPath("/node3/node31","node31".getBytes());
    }

6.4 查询ZNode节点

 @Test
    public void queryNode() throws Exception {
        //查询节点数据
        byte[] x = client.getData().forPath("/node1");
        System.out.println(new String(x));
        //查询子节点
        System.out.println(client.getChildren().forPath("/node3"));
        //更新节点
        client.setData().forPath("/node1","node1111".getBytes());
        //打印版本信息
        System.out.println(client.checkExists().forPath("/node1").getVersion());
 
        //获取子节点数据
        System.out.println(client.getChildren().forPath("/node3"));
        client.getChildren().inBackground(new BackgroundCallback() {
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println(curatorEvent.getChildren());
            }
        }).forPath("/node3");
 
    }

6.5 Transaction事务

@Test
    public void watcher() throws Exception {
        //开启事务
        client.inTransaction().create()
                .forPath("/node4")
                .and()
                .setData().forPath("/node5","node5".getBytes())//这个节点不存在，会抛异常
                .and()
                .commit();
    }

6.6 异步接口回调

@Test
    public void callback() throws Exception {
        client.create().inBackground(new BackgroundCallback() {
            public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
                System.out.println(curatorEvent.getType() + "   " + curatorEvent.getResultCode());
            }
        }, Executors.newFixedThreadPool(2))
                .forPath("/node5", "123".getBytes());
 
        while (true);
    }

6.7 分布式计数器

	@Test
    public void distributedAtomic() throws Exception {
        DistributedAtomicInteger atomicIntger =
                new DistributedAtomicInteger(client, "/count",
                        new RetryNTimes(3, 1000));
        AtomicValue<Integer> value = atomicIntger.add(1);
        System.out.println(value.succeeded());
        System.out.println(value.postValue());    //最新值
        System.out.println(value.preValue());    //原始值
    }

6.8 cache操作监听

Cache是一种缓存机制，可以借助Cache实现监听。

简单来说，cache在客户端缓存了znode的各种状态，当感知到zk集群的znode状态变化，会触发event事件，注册的监听器会处理这些事件。是不是很简单。

curator支持的cache种类有3种Path Cache，Node Cache，Tree Cache

1）Path Cache

Path Cache用来观察ZNode的子节点并缓存状态，如果ZNode的子节点被创建，更新或者删除，那么Path Cache会更新缓存，并且触发事件给注册的监听器。

Path Cache是通过PathChildrenCache类来实现的，监听器注册是通过PathChildrenCacheListener。

2）Node Cache

Node Cache用来观察ZNode自身，如果ZNode节点本身被创建，更新或者删除，那么Node Cache会更新缓存，并触发事件给注册的监听器。

Node Cache是通过NodeCache类来实现的，监听器对应的接口为NodeCacheListener。

3）Tree Cache

可以看做是上两种的合体，Tree Cache观察的是所有节点的所有数据。

//不可重入监听
    @Test
    public void watcher1() throws Exception {
 
        final NodeCache nodeCache = new NodeCache(client, "/node3/node31");
        NodeCacheListener nodeCacheListener = new NodeCacheListener() {
            public void nodeChanged() throws Exception {
                ChildData childData = nodeCache.getCurrentData();
                System.out.println("ZNode节点状态改变, path=" + childData.getPath());
                System.out.println("ZNode节点状态改变, data=" + childData.getData());
            }
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start();
        while (true);
    }
    //可重入监听
    @Test
    public void watcher2() throws Exception {
        TreeCache treeCache = new TreeCache(client, "/node3/node31");
        TreeCacheListener treeCacheListener = new TreeCacheListener() {
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData childData = event.getData();
                System.out.println("ZNode节点状态改变, path=" + childData.getPath());
                System.out.println("ZNode节点状态改变, data=" + childData.getData());
            }
        };
        treeCache.getListenable().addListener(treeCacheListener);
        treeCache.start();
        while (true);
    }