Zookeeper API及应用 – ZooKeeper教程™

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提供一个方法来处理znode所有使用ZooKeeper 集成。

客户端应遵循下面给出带 ZooKeeper 集成一个清晰的交互步骤。

  • 连接到ZooKeeper 。ZooKeeper 集成分配客户端的会话ID。
  • 定期发送心跳到服务器。否则,ZooKeeper 集成过期的会话ID,那么客户端需要重新连接。
  • 获得/设置只要znodes会话ID是活动的。
  • 从 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 − 一个执行对象“观察者”的接口。ZooKeeper 集合返回通过监控器对象的连接状态。

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

这里CountDownLatch用于停止(等待)主进程,直到客户端与 ZooKeeper 集合连接。

ZooKeeper集合通过观察回调回应连接状态。一旦客户与 ZooKeeper 集合连接,观察者回调调用CountDownLatch 释放锁倒计时方法在主进程中等待监视回调会被调用。

下面是完整的代码,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类提供了一个方法来在集合 ZooKeeper 创建一个新的 znode。创建方法的签名如下:

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

Where,

  • path − Znode的路径。例如 /myapp1, /myapp2, /myapp1/mydata1, myapp2/mydata1/myanothersubdata
  • data − 在一个指定的znode路径存储数据
  • acl − 要创建节点的访问控制列表。 ZooKeeperAPI提供了一个静态接口ZooDefs.Ids得到一些基本的ACL列表。例如,ZooDefs.Ids.OPEN_ACL_UNSAFE返回ACL开放的 znodes 列表。
  • createMode − 节点的类型,可以是短暂的,连续的,或两者。这是一个枚举类型。

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

连接方法将返回 ZooKeeper 对象 zk。 现在,调用自定义路径和数据创建 zk 对象的方法。.

完整的程序代码,以创建一个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
      }
   }
}

一旦应用程序编译和执行,使用指定数据的 znode 将在ZooKeeper集合创建。您可以使用 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 方法,创建 ZooKeeper 对象, “zk” 使用 “ZooKeeperConnection” 对象. 然后再调用zk” 对象的“exists”方法和指定的“path”。完整的列表如下 –

代码: 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类型的回调函数。ZooKeeper集合将通知通过观察者回调时指定的节点改变的数据。这是一次性的通知。
  • stat − 返回 znode 元数据。

让我们创建一个新的Java应用程序,以了解的ZooKeeperAPI的 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的数据将被改变,并且它可以使用 ZooKeeperCLI,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”类型函数. 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());
      }

   }

}

在运行程序之前,让我们使用 ZooKeeperCLI,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)

在这里,

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

让我们创建一个新的Java应用程序,以了解ZooKeeperAPI的删除功能。

创建一个文件 ZKDelete.java. 在main方法中,使用 ZooKeeperConnection对象创建一个 ZooKeeper 对象ZK。然后,调用 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
      }
   }
}

Zookeeper应用

Zookeeper 为分布式环境灵活协调提供基础架构。ZooKeeper框架支持许多当今许多工业应用。我们将在本章中讨论一些 ZooKeeper 显着的应用。

Yahoo!

ZooKeeper框架始建于“雅虎”。一个设计良好的分布式应用需求,满足需求,如数据的透明度,更好的性能,稳定性,集中配置和协调性。所以,他们设计了ZooKeeper框架来满足这些要求。

Apache Hadoop

Apache Hadoop是大数据行业的增长背后的驱动力。Hadoop依赖ZooKeeper的配置管理和协调。让我们举一个场景来了解 ZooKeeper 在 Hadoop 中的作用。

假设一个Hadoop集群桥接100以上的商用服务器。因此,有必要协调和命名这些服务器。由于大量节点的计算都参与其中,每个节点都需要互相同步,知道在哪里获得服务,并知道应该如何配置。在这个时间点上,Hadoop集群需要跨节点服务。ZooKeeper提供了跨节点同步的设备,确保整个的Hadoop项目的任务是序列化和同步的。

多重ZooKeeper服务器支持大型 Hadoop 集群。每个客户机与 ZooKeeper 的其中一个服务器通信,以检索和更新其同步信息。一些实时实例如下 –

  • 人类基因组计划 − 人类基因组计划中包含TB级别的数据。Hadoop的MapReduce框架可以用来分析数据集,并找到对人类发展有趣的事情。
  • 卫生保健 − 医院可以存储,检索和分析大量套门诊病历,通常都在TB级。

Apache HBase

Apache HBase是一个开源的,分布式的, 用于实时NoSQL数据库读取的大型数据集/写访问,并运行在HDFS之上。 HBase如下主从架构,HBase主管理所有的从服务。从服务被称为区域服务器。

HBase 分布式应用程序的安装取决于ZooKeeper运行的集群。Apache HBase 使用ZooKeeper 与集中配置管理,分布式互斥机制的帮助下跟踪分布式数据的整个主区域服务器的状态。下面是一些HBase的用例 –

  • 电信业 − 电信业存储数十亿的手机通话记录(大约30TB/月),访问这些通话记录实时成为一项艰巨的任务。HBase可用于所有记录的实时处理,容易且有效。
  • 社交网络 − 类似电信业,网站如Twitter,LinkedIn和Facebook通过用户创建的帖子内容接收大量数据。HBase 可以用来找出最近的趋势和其他有趣的事实。

Apache Solr

Apache Solr是用Java编写实现快速,开源的搜索平台。这是一个速度极快,容错分布式搜索引擎。建立在Lucene之上,它是一个高性能,全功能的文本搜索引擎。

Solr被广泛使用在ZooKeeper的每一个功能,如配置管理,领导人选举,节点管理,锁定和数据的同步。

Solr有两个不同的部分,索引和搜索。索引是将数据存储在一个适当的格式,以便它可以在以后检索处理。Solr使用ZooKeeper为索引数据在多个节点,并从多个节点中搜索。ZooKeeper 分布有以下特点-

  • 在需要的时候添加/删除节点
  • 最小化数据丢失并随后复制节点数据
  • 共享多个节点之间的数据,并从多个节点更快的搜索结果

一些Apache Solr实现用例包括电子商务,求职等。

未经允许不得转载:我爱IT聚乐部 » Zookeeper API及应用 – ZooKeeper教程™

打赏作者
很喜欢这篇文章,打赏犒劳一下作者,以此激励作者更多创作;

您的支持将鼓励我们继续创作!

[微信] 扫描二维码打赏

[支付宝] 扫描二维码打赏

赞 (0)
分享到:更多 ()

相关推荐

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址