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 集群将使会话 ID 过期，客户端需要重新连接。
只要会话 ID 有效，就可以获取/设置 znode。
完成所有任务后，断开与 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)

其中，

connectionString - ZooKeeper 集群主机。
sessionTimeout - 以毫秒为单位的会话超时时间。
watcher - 实现“Watcher”接口的对象。ZooKeeper 集群通过 watcher 对象返回连接状态。

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

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

ZooKeeper 集群通过Watcher 回调回复连接状态。一旦客户端连接到 ZooKeeper 集群，Watcher 回调将被调用，并且 Watcher 回调调用CountDownLatch的countDown方法以释放锁，主进程中的await。

以下是连接 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 集群。

创建 Znode

ZooKeeper 类提供create 方法以在 ZooKeeper 集群中创建一个新的 znode。create方法的签名如下：

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

其中，

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)

其中，

path - Znode 路径
watcher - 布尔值，用于指定是否监视指定的 znode

让我们创建一个新的 Java 应用程序来检查 ZooKeeper API 的“exists”功能。创建一个文件“ZKExists.java”。在 main 方法中，使用“ZooKeeperConnection”对象创建 ZooKeeper 对象“zk”。然后，使用自定义“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)

其中，

path - Znode 路径。
watcher - 类型为Watcher的回调函数。当指定 znode 的数据发生更改时，ZooKeeper 集群将通过 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)

其中，

path - Znode 路径
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)

其中，

path - Znode 路径。
watcher - 类型为“Watcher”的回调函数。当指定 znode 被删除或 znode 下的子节点被创建/删除时，ZooKeeper 集群将发出通知。这是一次性通知。

编码：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

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

myfirstsubnode
mysecondsubnode

删除 Znode

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

delete(String path, int version)

其中，

path - Znode 路径。
version - znode 的当前版本。

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

删除 znode 的完整程序代码如下：

编码：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
      }
   }
}

打印页面