---

Zookeeper - API

ZooKeeper 为 Java 和 C 提供了官方 API 绑定。ZooKeeper 社区为大多数语言（.NET、python 等）提供了非官方 API。 使用 ZooKeeper API，应用程序可以连接、交互、操作数据、协调并最终与 ZooKeeper 集合断开连接。

ZooKeeper API 具有一组丰富的功能，可以以简单安全的方式获得 ZooKeeper 集合的所有功能。 ZooKeeper API 提供同步和异步方法。

ZooKeeper

编码:和 ZooKeeper API 在各个方面完全互补，它极大地使开发人员受益。 让我们在本章中讨论 Java 绑定。

---

ZooKeeper API 基础

与 ZooKeeper 集合交互的应用程序称为 ZooKeeper 客户端或简称为客户端。

Znode 是 ZooKeeper 集合的核心组件，ZooKeeper API 提供了一小部分方法来使用 ZooKeeper 集合操作 znode 的所有细节。

客户端应该按照下面给出的步骤与 ZooKeeper 集合进行清晰和干净的交互。

• 连接到 ZooKeeper 集合。 ZooKeeper 集合为客户端分配一个会话 ID。

• 定期向服务器发送心跳。 否则，ZooKeeper 集合的 Session ID 到期，客户端需要重新连接。

• 只要会话 ID 处于活动状态，就获取 / 设置 znodes。

• 完成所有任务后，从 ZooKeeper 集 断开连接。 如果客户端长时间不活动，ZooKeeper 集合将自动断开客户端。

---

Java 绑定

让我们了解本章中最重要的一组 ZooKeeper API。 ZooKeeper API 的核心部分是 ZooKeeper 类。 它在其构造函数中提供了连接 ZooKeeper 集合的选项，并具有以下方法 −

• connect − 连接到 ZooKeeper 集合

• create − 创建一个znode

• exists − 检查znode是否存在及其信息

• getData − 从特定 znode 获取数据

• setData − 在特定的 znode 中设置数据

• getChildren − 获取特定 znode 中可用的所有子节点

• delete − 获取特定的 znode 及其所有子节点

• close − 关闭连接

---

连接到 ZooKeeper 集合

ZooKeeper 类通过其构造函数提供连接功能。 构造函数的签名如下 −

ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

Where,

• connectionString − ZooKeeper 集合主机。

• sessionTimeout − 会话超时以毫秒为单位。

• watcher − 一个实现"监听"接口的对象。 ZooKeeper 集合通过 watcher 对象返回连接状态。

让我们创建一个新的帮助类 ZooKeeperConnection 并添加一个方法 connect。 connect 方法创建一个 ZooKeeper 对象，连接到 ZooKeeper 集合，然后返回该对象。

这里的 CountDownLatch 用于停止（等待）主进程，直到客户端连接到 ZooKeeper 集合。

ZooKeeper 集合通过监听回调回复连接状态。一旦客户端与 ZooKeeper 集合连接，监听回调就会被调用，监听回调会调用 CountDownLatch 的 countDown 方法来释放锁，在主进程中等待。

这是与 ZooKeeper 集合连接的完整代码。

编码: ZooKeeperConnection.java

// import java classes
import java.io.IOException;
import java.util.concurrent.CountDownLatch;

// import zookeeper classes
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat;

public class ZooKeeperConnection {

   // declare zookeeper instance to access ZooKeeper ensemble
   private ZooKeeper zoo;
   final CountDownLatch connectedSignal = new CountDownLatch(1);

   // Method to connect zookeeper ensemble.
   public ZooKeeper connect(String host) throws IOException,InterruptedException {
	
      zoo = new ZooKeeper(host,5000,new Watcher() {
		
         public void process(WatchedEvent we) {

            if (we.getState() == KeeperState.SyncConnected) {
               connectedSignal.countDown();
            }
         }
      });
		
      connectedSignal.await();
      return zoo;
   }

   // Method to disconnect from zookeeper server
   public void close() throws InterruptedException {
      zoo.close();
   }
}

保存上面的代码，它将在下一节中用于连接 ZooKeeper ensemble。

---

创建一个 Znode

ZooKeeper 类提供了 create 方法 来在 ZooKeeper 集合中创建一个新的 znode。 create 方法的签名如下 −

create(String path, byte[] data, List<ACL> acl, CreateMode createMode)

Where,

• path − Znode 路径。 例如，/myapp1、/myapp2、/myapp1/mydata1、myapp2/mydata1/myanothersubdata

• data − 要存储在指定 znode 路径中的数据

• acl − 要创建的节点的访问控制列表。 ZooKeeper API 提供了一个静态接口ZooDefs.Ids 来获取一些基本的acl 列表。 例如，ZooDefs.Ids.OPEN_ACL_UNSAFE 返回打开 znode 的 acl 列表。

• createMode − 节点的类型，可以是短暂的、顺序的，或者两者兼而有之。 这是一个枚举。

让我们创建一个新的 Java 应用程序来检查 ZooKeeper API 的 create 功能。 创建一个文件 ZKCreate.java。 在 main 方法中，创建一个 ZooKeeperConnection 类型的对象并调用 connect 方法以连接到 ZooKeeper 集合。

connect 方法将返回 ZooKeeper 对象 zk。 现在，使用自定义的 path 和 data 调用 zk 对象的 create 方法。

创建znode的完整程序代码如下 −

编码: ZKCreate.java

import java.io.IOException;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

public class ZKCreate {
   // create static instance for zookeeper class.
   private static ZooKeeper zk;

   // create static instance for ZooKeeperConnection class.
   private static ZooKeeperConnection conn;

   // Method to create znode in zookeeper ensemble
   public static void create(String path, byte[] data) throws 
      KeeperException,InterruptedException {
      zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
      CreateMode.PERSISTENT);
   }

   public static void main(String[] args) {

      // znode path
      String path = "/MyFirstZnode"; // Assign path to znode

      // data in byte array
      byte[] data = "My first zookeeper app”.getBytes(); // Declare data
		
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         create(path, data); // Create the data to the specified path
         conn.close();
      } catch (Exception e) {
         System.out.println(e.getMessage()); //Catch error message
      }
   }
}

编译并执行应用程序后，将在 ZooKeeper 集合中创建具有指定数据的 znode。 您可以使用 ZooKeeper CLI zkCli.sh 检查它。

cd /path/to/zookeeper
bin/zkCli.sh
>>> get /MyFirstZnode

---

Exists – 检查 Znode 的存在

ZooKeeper 类提供了 exists 方法 来检查 znode 的存在。 如果指定的 znode 存在，则返回 znode 的元数据。 exists 方法的签名如下 −

exists(String path, boolean watcher)

Where,

• path − Znode 路径

• watcher − 布尔值，用于指定是否监视指定的 znode

让我们创建一个新的 Java 应用程序来检查 ZooKeeper API 的"exists"创建一个文件"ZKExists.java"。在 main 方法中，创建 ZooKeeper 对象，"zk" 使用 "ZooKeeperConnection" 对象。然后，用自定义的"path"调用"zk"对象的"exists"方法。完整的列表如下 −

编码: ZKExists.java

import java.io.IOException;

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;

public class ZKExists {
   private static ZooKeeper zk;
   private static ZooKeeperConnection conn;

   // Method to check existence of znode and its status, if znode is available.
   public static Stat znode_exists(String path) throws
      KeeperException,InterruptedException {
      return zk.exists(path, true);
   }

   public static void main(String[] args) throws InterruptedException,KeeperException {
      String path = "/MyFirstZnode"; // Assign znode to the specified path
			
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         Stat stat = znode_exists(path); // Stat checks the path of the znode
				
         if(stat != null) {
            System.out.println("Node exists and the node version is " +
            stat.getVersion());
         } else {
            System.out.println("Node does not exists");
         }
				
      } catch(Exception e) {
         System.out.println(e.getMessage()); // Catches error messages
      }
   }
}

编译并执行应用程序后，您将获得以下输出。

Node exists and the node version is 1.

---

getData 方法

ZooKeeper 类提供 getData 方法来获取附加在指定 znode 中的数据及其状态。 getData 方法的签名如下 −

getData(String path, Watcher watcher, Stat stat)

Where,

• path − Znode 路径。

• watcher − Watcher 类型的回调函数。 ZooKeeper

  编码:当指定znode的数据发生变化时，会通过Watcher回调通知。 这是一次性通知。

• stat − 返回 znode 的元数据。

让我们创建一个新的 Java 应用程序来了解 ZooKeeper API 的 getData 功能。创建一个文件 ZKGetData.java。 在 main 方法中，使用 ZooKeeperConnection 对象创建一个 ZooKeeper 对象 zk。然后，使用自定义路径调用 zk 对象的 getData 方法。

这是从指定节点获取数据的完整程序代码 −

编码: ZKGetData.java

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

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;

public class ZKGetData {

   private static ZooKeeper zk;
   private static ZooKeeperConnection conn;
   public static Stat znode_exists(String path) throws 
      KeeperException,InterruptedException {
      return zk.exists(path,true);
   }

   public static void main(String[] args) throws InterruptedException, KeeperException {
      String path = "/MyFirstZnode";
      final CountDownLatch connectedSignal = new CountDownLatch(1);
		
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         Stat stat = znode_exists(path);
			
         if(stat != null) {
            byte[] b = zk.getData(path, new Watcher() {
				
               public void process(WatchedEvent we) {
					
                  if (we.getType() == Event.EventType.None) {
                     switch(we.getState()) {
                        case Expired:
                        connectedSignal.countDown();
                        break;
                     }
							
                  } else {
                     String path = "/MyFirstZnode";
							
                     try {
                        byte[] bn = zk.getData(path,
                        false, null);
                        String data = new String(bn,
                        "UTF-8");
                        System.out.println(data);
                        connectedSignal.countDown();
							
                     } catch(Exception ex) {
                        System.out.println(ex.getMessage());
                     }
                  }
               }
            }, null);
				
            String data = new String(b, "UTF-8");
            System.out.println(data);
            connectedSignal.await();
				
         } else {
            System.out.println("Node does not exists");
         }
      } catch(Exception e) {
        System.out.println(e.getMessage());
      }
   }
}

编译并执行应用程序后，您将获得以下输出

My first zookeeper app

并且应用程序将等待来自 ZooKeeper 集合的进一步通知。 使用 ZooKeeper CLI zkCli.sh 更改指定 znode 的数据。

cd /path/to/zookeeper
bin/zkCli.sh
>>> set /MyFirstZnode Hello

现在，应用程序将打印以下输出并退出。

Hello

---

setData 方法

ZooKeeper 类提供 setData 方法来修改附加在指定 znode 中的数据。 setData 方法的签名如下 −

setData(String path, byte[] data, int version)

Where,

• path − 路径

• data − 要存储在指定 znode 路径中的数据。

• version − znode 的当前版本。 每当数据发生变化时，ZooKeeper 都会更新 znode 的版本号。

现在让我们创建一个新的 Java 应用程序来了解 ZooKeeper API 的 setData 功能。创建一个文件 ZKSetData.java。 在 main 方法中，使用 ZooKeeperConnection 对象创建一个 ZooKeeper 对象 zk。然后，调用 zk 对象的 setData 方法，指定路径、新数据和节点版本。

这是修改指定znode中附加的数据的完整程序代码。

代码: ZKSetData.java

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;

import java.io.IOException;

public class ZKSetData {
   private static ZooKeeper zk;
   private static ZooKeeperConnection conn;

   // Method to update the data in a znode. Similar to getData but without watcher.
   public static void update(String path, byte[] data) throws
      KeeperException,InterruptedException {
      zk.setData(path, data, zk.exists(path,true).getVersion());
   }

   public static void main(String[] args) throws InterruptedException,KeeperException {
      String path= "/MyFirstZnode";
      byte[] data = "Success".getBytes(); //Assign data which is to be updated.
		
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         update(path, data); // Update znode data to the specified path
      } catch(Exception e) {
         System.out.println(e.getMessage());
      }
   }
}

一旦应用程序被编译和执行，指定znode的数据将被改变，并且可以使用ZooKeeper CLI，zkCli.sh来检查。

cd /path/to/zookeeper
bin/zkCli.sh
>>> get /MyFirstZnode

---

getChildren 方法

ZooKeeper 类提供 getChildren 方法来获取特定 znode 的所有子节点。 getChildren 方法的签名如下 −

getChildren(String path, Watcher watcher)

Where,

• path − Znode 路径。

• watcher − "Watcher"类型的回调函数。ZooKeeper

  编码:将在指定的 znode 被删除或 znode 下的子节点被创建/删除时通知。 这是一次性通知。

编码: ZKGetChildren.java

import java.io.IOException;
import java.util.*;

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;

public class ZKGetChildren {
   private static ZooKeeper zk;
   private static ZooKeeperConnection conn;

   // Method to check existence of znode and its status, if znode is available.
   public static Stat znode_exists(String path) throws 
      KeeperException,InterruptedException {
      return zk.exists(path,true);
   }

   public static void main(String[] args) throws InterruptedException,KeeperException {
      String path = "/MyFirstZnode"; // Assign path to the znode
		
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         Stat stat = znode_exists(path); // Stat checks the path

         if(stat!= null) {

            //“getChildren” method- get all the children of znode.It has two
            args, path and watch
            List <String> children = zk.getChildren(path, false);
            for(int i = 0; i < children.size(); i++)
            System.out.println(children.get(i)); //Print children's
         } else {
            System.out.println("Node does not exists");
         }

      } catch(Exception e) {
         System.out.println(e.getMessage());
      }

   }

}

在运行程序之前，让我们使用 ZooKeeper CLI zkCli.sh 为 /MyFirstZnode 创建两个子节点。

cd /path/to/zookeeper
bin/zkCli.sh
>>> create /MyFirstZnode/myfirstsubnode Hi
>>> create /MyFirstZnode/mysecondsubmode Hi

现在，编译并运行程序将输出上面创建的 znodes。

myfirstsubnode
mysecondsubnode

---

删除一个 Znode

ZooKeeper 类提供了 delete 方法来删除指定的 znode。 delete 方法的签名如下 −

delete(String path, int version)

Where,

• path − Znode 路径。

• version − znode 的当前版本。

让我们创建一个新的 Java 应用程序来了解 ZooKeeper API 的 delete 功能。创建一个文件 ZKDelete.java。在 main 方法中，使用 ZooKeeperConnection 对象创建一个 ZooKeeper 对象 zk。然后，使用指定的path和节点的版本调用zk对象的delete方法。

The complete program code to delete a znode is as follows −

编码: ZKDelete.java

import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException;

public class ZKDelete {
   private static ZooKeeper zk;
   private static ZooKeeperConnection conn;

   // Method to check existence of znode and its status, if znode is available.
   public static void delete(String path) throws KeeperException,InterruptedException {
      zk.delete(path,zk.exists(path,true).getVersion());
   }

   public static void main(String[] args) throws InterruptedException,KeeperException {
      String path = "/MyFirstZnode"; //Assign path to the znode
		
      try {
         conn = new ZooKeeperConnection();
         zk = conn.connect("localhost");
         delete(path); //delete the node with the specified path
      } catch(Exception e) {
         System.out.println(e.getMessage()); // catches error messages
      }
   }
}

---