📅  最后修改于: 2020-11-28 14:27:49             🧑  作者: Mango
ZooKeeper具有Java和C的正式API绑定。ZooKeeper社区为大多数语言(.NET, Python等)提供了非官方的API。使用ZooKeeper API,应用程序可以连接,交互,操纵数据,协调并最终与ZooKeeper集成断开连接。
ZooKeeper API具有丰富的功能集,以简单安全的方式获得ZooKeeper集成的所有功能。 ZooKeeper API提供同步和异步方法。
ZooKeeper集成和ZooKeeper API在各个方面都完全互补,并且极大地有益于开发人员。让我们在本章中讨论Java绑定。
与ZooKeeper集成交互的应用程序称为ZooKeeper客户端或简称为Client 。
Znode是ZooKeeper集成的核心组件,而ZooKeeper API提供了一些方法来使用ZooKeeper集成来操纵znode的所有细节。
客户应按照以下给出的步骤与ZooKeeper集成进行清晰,干净的交互。
连接到ZooKeeper合奏。 ZooKeeper集成为客户端分配会话ID。
定期向服务器发送心跳。否则,ZooKeeper集合将使会话ID过期,并且客户端需要重新连接。
只要会话ID处于活动状态,就获取/设置znodes。
完成所有任务后,请与ZooKeeper集成断开连接。如果客户端长时间不活动,则ZooKeeper集成将自动断开客户端的连接。
让我们了解本章中最重要的ZooKeeper API集。 ZooKeeper API的中心部分是ZooKeeper类。它提供了在其构造函数中连接ZooKeeper集成的选项,并具有以下方法-
连接-连接到ZooKeeper合奏
创建-创建一个znode
存在-检查znode是否存在及其信息
的getData -从特定Z序节点获取数据
使用setData -在一个特定的Z序节点设定数据
getChildren-获取特定znode中所有可用的子节点
删除-获取特定的znode及其所有子节点
关闭-关闭连接
ZooKeeper类通过其构造函数提供连接功能。构造函数的签名如下-
ZooKeeper(String connectionString, int sessionTimeout, Watcher watcher)
哪里,
connectionString -ZooKeeper合奏主机。
sessionTimeout-会话超时(以毫秒为单位)。
watcher-一个实现“ Watcher”接口的对象。 ZooKeeper集合通过观察者对象返回连接状态。
让我们创建一个新的帮助程序类ZooKeeperConnection并添加一个方法connect 。 connect方法创建一个ZooKeeper对象,连接到ZooKeeper集合,然后返回该对象。
在这里, CountDownLatch用于停止(等待)主进程,直到客户端与ZooKeeper集成连接为止。
ZooKeeper集成通过Watcher回调回复连接状态。客户端与ZooKeeper集成连接后,将调用Watcher回调,并且Watcher回调调用CountDownLatch的countDown方法释放锁,并在主进程中等待。
这是连接ZooKeeper集成的完整代码。
// 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集成。
ZooKeeper类提供了create方法,以在ZooKeeper集成中创建新的znode。 create方法的签名如下-
create(String path, byte[] data, List acl, CreateMode createMode)
哪里,
path -Znode路径。例如,/ myapp1,/ myapp2,/ myapp1 / mydata1,myapp2 / mydata1 / myanothersubdata
数据-数据存储在指定的Z序节点路径
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 。现在,使用自定义路径和data调用zk对象的create方法。
创建znode的完整程序代码如下-
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
ZooKeeper类提供了现存的方法来检查znode的存在。如果指定的znode存在,它将返回znode的元数据。存在方法的签名如下-
exists(String path, boolean watcher)
哪里,
路径-Znode路径
观察家-布尔值,指定是否要观看特定Z序节点或不
让我们创建一个新的Java应用程序来检查ZooKeeper API的“存在”功能。创建一个文件“ ZKExists.java” 。在主要方法中,使用“ ZooKeeperConnection”对象创建ZooKeeper对象“ zk” 。然后,使用自定义的“路径”调用“ zk”对象的“存在”方法。完整的清单如下-
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.
ZooKeeper类提供了getData方法来获取附加到指定znode中的数据及其状态。 getData方法的签名如下-
getData(String path, Watcher watcher, Stat stat)
哪里,
path -Znode路径。
watcher-类型Watcher的回调函数。当指定的znode的数据发生更改时,ZooKeeper集合将通过Watcher回调进行通知。这是一次性通知。
统计-返回Z序节点的元数据。
让我们创建一个新的Java应用程序以了解ZooKeeper API的getData功能。创建一个文件ZKGetData.java 。在main方法中,使用ZooKeeperConnection对象创建一个ZooKeeper对象zk 。然后,使用自定义路径调用zk对象的getData方法。
这是从指定节点获取数据的完整程序代码-
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
ZooKeeper类提供setData方法来修改附加到指定znode中的数据。 setData方法的签名如下-
setData(String path, byte[] data, int version)
哪里,
路径-Znode路径
数据-要存储在指定znode路径中的数据。
版本-znode的当前版本。每当更改数据时,ZooKeeper都会更新znode的版本号。
现在让我们创建一个新的Java应用程序以了解ZooKeeper API的setData功能。创建一个文件ZKSetData.java 。在main方法中,使用ZooKeeperConnection对象创建一个ZooKeeper对象zk 。然后,使用指定的路径,新数据和节点版本调用zk对象的setData方法。
这是修改指定znode中附加数据的完整程序代码。
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
ZooKeeper类提供了getChildren方法来获取特定znode的所有子节点。 getChildren方法的签名如下-
getChildren(String path, Watcher watcher)
哪里,
path -Znode路径。
watcher-类型为“ Watcher”的回调函数。 ZooKeeper合奏将通知删除指定的znode或创建/删除znode下的子级。这是一次性通知。
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 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
ZooKeeper类提供了delete方法来删除指定的znode。 delete方法的签名如下-
delete(String path, int version)
哪里,
path -Znode路径。
版本-znode的当前版本。
让我们创建一个新的Java应用程序以了解ZooKeeper API的删除功能。创建一个文件ZKDelete.java 。在main方法中,使用ZooKeeperConnection对象创建一个ZooKeeper对象zk 。然后,使用节点的指定路径和版本调用zk对象的delete方法。
删除znode的完整程序代码如下-
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
}
}
}