返回介绍
文江博客 开发文档 ZooKeeper 教程 文章详情

ZooKeeper API

发布于 2024-06-23 18:43:40 字数 32646 浏览 0 评论 0 收藏 0

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

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

ZooKeeper集群和ZooKeeper API在各个方面都完全互补，并且极大地有益于开发人员。让我们在本章中讨论Java绑定。

ZooKeeper API的基础

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

Znode是ZooKeeper集群的核心组件，而ZooKeeper API提供了一组方法来使用ZooKeeper集群来操纵znode的所有细节。

客户端API应遵循以下给出的步骤，与ZooKeeper集群团进行清晰，干净的交互。

  • 连接到ZooKeeper集群。ZooKeeper集群为客户端分配会话ID。
  • 定期向服务器发送心跳。否则，ZooKeeper集群将使会话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 类通过其构造函数提供连接功能。构造函数的签名如下：

  1. ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)

参数：

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

让我们创建ZooKeeperConnection类并添加一个方法connect。该连接方法创建一个ZooKeeper对象，所连接到的ZooKeeper集群，然后返回该对象。在这里，CountDownLatch用于停止（等待）主进程，直到客户端与ZooKeeper集群连接为止。 ZooKeeper集群通过Watcher回调返回连接状态。客户端与ZooKeeper集群连接后，将调用Watcher回调，并且Watcher回调调用CountDownLatch的countDown方法释放锁，并在主进程中等待。

下面这是连接ZooKeeper集群的完整代码。

  1. //import java classes
  2. import java.io.IOException;
  3. import java.util.concurrent.CountDownLatch;
  5. //import zookeeper classes
  6. import org.apache.zookeeper.KeeperException;
  7. import org.apache.zookeeper.WatchedEvent;
  8. import org.apache.zookeeper.Watcher;
  9. import org.apache.zookeeper.Watcher.Event.KeeperState;
  10. import org.apache.zookeeper.ZooKeeper;
  11. import org.apache.zookeeper.AsyncCallback.StatCallback;
  12. import org.apache.zookeeper.KeeperException.Code;
  13. import org.apache.zookeeper.data.Stat;
  15. public class ZooKeeperConnection {
  17.    //declare zookeeper instance to access ZooKeeper ensemble
  18.    private ZooKeeper zoo;
  19.    final CountDownLatch connectedSignal = new CountDownLatch(1);
  21.    //Method to connect zookeeper ensemble.
  22.    public ZooKeeper connect(String host) throws IOException,InterruptedException {
  24.       zoo = new ZooKeeper(host,5000,new Watcher() {
  26.          public void process(WatchedEvent we) {
  28.             if (we.getState() == KeeperState.SyncConnected) {
  29.                connectedSignal.countDown();
  30.             }
  31.          }
  32.       });
  34.       connectedSignal.await();
  35.       return zoo;
  36.    }
  38.    //Method to disconnect from zookeeper server
  39.    public void close() throws InterruptedException {
  40.       zoo.close();
  41.    }
  42. }

保存以上代码，该代码将在下一部分中用于连接ZooKeeper集群。

创建一个Znode

ZooKeeper类提供了create方法，以在ZooKeeper集成中创建新的znode。create方法的用法如下：

  1. 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 的创建功能。创建一个文件ZKCreate.java。在main方法中，创建类型为ZooKeeperConnection的对象，并调用connect方法以连接到ZooKeeper集群。 connect方法将返回ZooKeeper对象zk。现在，使用自定义路径和数据调用zk对象的create方法。创建znode的完整程序代码如下：

ZKCreate.java

  1. import java.io.IOException;
  3. import org.apache.zookeeper.WatchedEvent;
  4. import org.apache.zookeeper.Watcher;
  5. import org.apache.zookeeper.Watcher.Event.KeeperState;
  6. import org.apache.zookeeper.ZooKeeper;
  7. import org.apache.zookeeper.KeeperException;
  8. import org.apache.zookeeper.CreateMode;
  9. import org.apache.zookeeper.ZooDefs;
  11. public class ZKCreate {
  12.    // create static instance for zookeeper class.
  13.    private static ZooKeeper zk;
  15.    // create static instance for ZooKeeperConnection class.
  16.    private static ZooKeeperConnection conn;
  18.    // Method to create znode in zookeeper ensemble
  19.    public static void create(String path, byte[] data) throws 
  20.       KeeperException,InterruptedException {
  21.       zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE,
  22.       CreateMode.PERSISTENT);
  23.    }
  25.    public static void main(String[] args) {
  27.       // znode path
  28.       String path = "/MyFirstZnode"; // Assign path to znode
  30.       // data in byte array
  31.       byte[] data = "My first zookeeper app”.getBytes(); // Declare data
  33.       try {
  34.          conn = new ZooKeeperConnection();
  35.          zk = conn.connect("localhost");
  36.          create(path, data); // Create the data to the specified path
  37.          conn.close();
  38.       } catch (Exception e) {
  39.          System.out.println(e.getMessage()); //Catch error message
  40.       }
  41.    }
  42. }

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

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

exists：检查Znode的存在

ZooKeeper类提供了exists方法来检查znode的存在。如果指定的znode存在，它将返回znode的元数据。exists方法的用法如下：
语法：

  1. exists(String path, boolean watcher)

参数：

  • path：Znode路径
  • watcher：布尔值，指定是否要观看特定Z序节点或不看

让我们创建一个新的Java应用程序来检查ZooKeeper API的“exists”功能。创建一个文件“ ZKExists.java”。在主要方法中，使用“ZooKeeperConnection”对象创建ZooKeeper对象“zk”。然后，使用自定义“path”调用“zk”对象的“ exists”方法。完整的清单如下：
ZKExists.java

  1. import java.io.IOException;
  3. import org.apache.zookeeper.ZooKeeper;
  4. import org.apache.zookeeper.KeeperException;
  5. import org.apache.zookeeper.WatchedEvent;
  6. import org.apache.zookeeper.Watcher;
  7. import org.apache.zookeeper.Watcher.Event.KeeperState;
  8. import org.apache.zookeeper.data.Stat;
  10. public class ZKExists {
  11.    private static ZooKeeper zk;
  12.    private static ZooKeeperConnection conn;
  14.    // Method to check existence of znode and its status, if znode is available.
  15.    public static Stat znode_exists(String path) throws
  16.       KeeperException,InterruptedException {
  17.       return zk.exists(path, true);
  18.    }
  20.    public static void main(String[] args) throws InterruptedException,KeeperException {
  21.       String path = "/MyFirstZnode"; // Assign znode to the specified path
  23.       try {
  24.          conn = new ZooKeeperConnection();
  25.          zk = conn.connect("localhost");
  26.          Stat stat = znode_exists(path); // Stat checks the path of the znode
  28.          if(stat != null) {
  29.             System.out.println("Node exists and the node version is " +
  30.             stat.getVersion());
  31.          } else {
  32.             System.out.println("Node does not exists");
  33.          }
  35.       } catch(Exception e) {
  36.          System.out.println(e.getMessage()); // Catches error messages
  37.       }
  38.    }
  39. }

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

  1. Node exists and the node version is 1.

getData方法

ZooKeeper类提供了getData方法来获取附加到指定znode中的数据及其状态。getData方法的用法如下：
语法：

  1. 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

  1. import java.io.IOException;
  2. import java.util.concurrent.CountDownLatch;
  4. import org.apache.zookeeper.ZooKeeper;
  5. import org.apache.zookeeper.KeeperException;
  6. import org.apache.zookeeper.WatchedEvent;
  7. import org.apache.zookeeper.Watcher;
  8. import org.apache.zookeeper.Watcher.Event.KeeperState;
  9. import org.apache.zookeeper.data.Stat;
  11. public class ZKGetData {
  13.    private static ZooKeeper zk;
  14.    private static ZooKeeperConnection conn;
  15.    public static Stat znode_exists(String path) throws 
  16.       KeeperException,InterruptedException {
  17.       return zk.exists(path,true);
  18.    }
  20.    public static void main(String[] args) throws InterruptedException, KeeperException {
  21.       String path = "/MyFirstZnode";
  22.       final CountDownLatch connectedSignal = new CountDownLatch(1);
  24.       try {
  25.          conn = new ZooKeeperConnection();
  26.          zk = conn.connect("localhost");
  27.          Stat stat = znode_exists(path);
  29.          if(stat != null) {
  30.             byte[] b = zk.getData(path, new Watcher() {
  32.                public void process(WatchedEvent we) {
  34.                   if (we.getType() == Event.EventType.None) {
  35.                      switch(we.getState()) {
  36.                         case Expired:
  37.                         connectedSignal.countDown();
  38.                         break;
  39.                      }
  40.                   } else {
  41.                      String path = "/MyFirstZnode";
  42.                      try {
  43.                         byte[] bn = zk.getData(path,
  44.                         false, null);
  45.                         String data = new String(bn,
  46.                         "UTF-8");
  47.                         System.out.println(data);
  48.                         connectedSignal.countDown();
  50.                      } catch(Exception ex) {
  51.                         System.out.println(ex.getMessage());
  52.                      }
  53.                   }
  54.                }
  55.             }, null);
  57.             String data = new String(b, "UTF-8");
  58.             System.out.println(data);
  59.             connectedSignal.await();
  61.          } else {
  62.             System.out.println("Node does not exists");
  63.          }
  64.       } catch(Exception e) {
  65.         System.out.println(e.getMessage());
  66.       }
  67.    }
  68. }

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

  1. My first zookeeper app

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

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

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

  1. Hello

setData方法

ZooKeeper类提供setData方法来修改附加到指定znode中的数据。setData方法的用法如下：

  1. 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

  1. import org.apache.zookeeper.ZooKeeper;
  2. import org.apache.zookeeper.KeeperException;
  3. import org.apache.zookeeper.WatchedEvent;
  4. import org.apache.zookeeper.Watcher;
  5. import org.apache.zookeeper.Watcher.Event.KeeperState;
  7. import java.io.IOException;
  9. public class ZKSetData {
  10.    private static ZooKeeper zk;
  11.    private static ZooKeeperConnection conn;
  13.    // Method to update the data in a znode. Similar to getData but without watcher.
  14.    public static void update(String path, byte[] data) throws
  15.       KeeperException,InterruptedException {
  16.       zk.setData(path, data, zk.exists(path,true).getVersion());
  17.    }
  19.    public static void main(String[] args) throws InterruptedException,KeeperException {
  20.       String path= "/MyFirstZnode";
  21.       byte[] data = "Success".getBytes(); //Assign data which is to be updated.
  23.       try {
  24.          conn = new ZooKeeperConnection();
  25.          zk = conn.connect("localhost");
  26.          update(path, data); // Update znode data to the specified path
  27.       } catch(Exception e) {
  28.          System.out.println(e.getMessage());
  29.       }
  30.    }
  31. }

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

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

getChildren方法

ZooKeeper类提供了getChildren方法来获取特定znode的所有子节点。getChildren方法的用法如下：

  1. getChildren(String path, Watcher watcher)

参数：

  • path：Znode路径。
  • watcher：类型为Watcher的回调函数。当指定的znode的数据发生更改时，ZooKeeper集群将通过Watcher回调进行通知。这是一次性通知。

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

这是从指定节点获取数据的完整程序代码：
ZKGetChildren.java

  1. import java.io.IOException;
  2. import java.util.*;
  4. import org.apache.zookeeper.ZooKeeper;
  5. import org.apache.zookeeper.KeeperException;
  6. import org.apache.zookeeper.WatchedEvent;
  7. import org.apache.zookeeper.Watcher;
  8. import org.apache.zookeeper.Watcher.Event.KeeperState;
  9. import org.apache.zookeeper.data.Stat;
  11. public class ZKGetChildren {
  12.    private static ZooKeeper zk;
  13.    private static ZooKeeperConnection conn;
  15.    // Method to check existence of znode and its status, if znode is available.
  16.    public static Stat znode_exists(String path) throws 
  17.       KeeperException,InterruptedException {
  18.       return zk.exists(path,true);
  19.    }
  21.    public static void main(String[] args) throws InterruptedException,KeeperException {
  22.       String path = "/MyFirstZnode"; // Assign path to the znode
  24.       try {
  25.          conn = new ZooKeeperConnection();
  26.          zk = conn.connect("localhost");
  27.          Stat stat = znode_exists(path); // Stat checks the path
  29.          if(stat!= null) {
  31.             //“getChildren” method- get all the children of znode.It has two
  32.             args, path and watch
  33.             List <String> children = zk.getChildren(path, false);
  34.             for(int i = 0; i < children.size(); i++)
  35.             System.out.println(children.get(i)); //Print children's
  36.          } else {
  37.             System.out.println("Node does not exists");
  38.          }
  40.       } catch(Exception e) {
  41.          System.out.println(e.getMessage());
  42.       }
  43.    }
  44. }

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

  1. cd /path/to/zookeeper
  2. bin/zkCli.sh
  3. >>> create /MyFirstZnode/myfirstsubnode Hi
  4. >>> create /MyFirstZnode/mysecondsubmode Hi

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

  1. myfirstsubnode
  2. mysecondsubnode

删除Znode

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

  1. delete(String path, int version)

参数：

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

让我们创建一个新的Java应用程序以了解ZooKeeper API 的删除功能。创建一个文件ZKDelete.java。在main方法中，使用ZooKeeperConnection对象创建一个ZooKeeper对象zk。然后，使用节点的指定路径和版本调用zk对象的delete方法。删除znode的完整程序代码如下：
ZKDelete.java

  1. import org.apache.zookeeper.ZooKeeper;
  2. import org.apache.zookeeper.KeeperException;
  4. public class ZKDelete {
  5.    private static ZooKeeper zk;
  6.    private static ZooKeeperConnection conn;
  8.    // Method to check existence of znode and its status, if znode is available.
  9.    public static void delete(String path) throws KeeperException,InterruptedException {
  10.       zk.delete(path,zk.exists(path,true).getVersion());
  11.    }
  13.    public static void main(String[] args) throws InterruptedException,KeeperException {
  14.       String path = "/MyFirstZnode"; //Assign path to the znode
  16.       try {
  17.          conn = new ZooKeeperConnection();
  18.          zk = conn.connect("localhost");
  19.          delete(path); //delete the node with the specified path
  20.       } catch(Exception e) {
  21.          System.out.println(e.getMessage()); // catches error messages
  22.       }
  23.    }
  24. }

如果你对这篇内容有疑问，欢迎到本站社区发帖提问 参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

扫码二维码加入Web技术交流群

发布评论

需要 登录 才能够评论， 你可以免费 注册 一个本站的账号。
列表为空，暂无数据
    我们使用 Cookies 和其他技术来定制您的体验包括您的登录状态等。通过阅读我们的 隐私政策 了解更多相关信息。 单击 接受 或继续使用网站，即表示您同意使用 Cookies 和您的相关数据。
    原文