Java调用Zookeeper的实现步骤

Java调用Zookeeper的实现步骤

目录watch机制常用APIjava调用

watch机制

Zookeeper watch是一种监听通知机制，可以随时监听一些数据的变化，从而实现数据的及时性。

Zookeeper所有的读操作getData(), getChildren()和 exists()都可以设置监听(watch)。【写操作则是不能设置监视点的。】

Watch的三个关键点：

一次有效：当设置监视的数据发生改变时，该监视事件会被发送到客户端，并且该监听将会停止，除非重启注册监听；

顺序保证：网络延迟或者其他因素可能导致不同的客户端在不同的时刻感知某一监视事件，但是不同的客户端所看到的一切具有一致的顺序；

可以监听数据和子节点：getData()和 exists()可以设置监听数据变化；getChildren 可以设置监听子节点变化；

常用API

/**

*　构造器

* @param connectString 集群的IP：端口号；多个服务器时，中间用逗号分割

* @param sessionTimeout 超时时间，单位：毫秒

* @param watcher 监听器，监听节点变化

* @throws IOException

*/

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException

/**

*

* @param path 节点路径

* @param data 数据

* @param acl 访问控制列表

* @param createMode 节点类型

* @return

* @throws KeeperException

* @throws InterruptedException

*/

public String create(String path, byte[] data, List acl, CreateMode createMode) throws KeeperException, InterruptedException

/**

*

* @param path 节点路径

* @param watch 监听器

* @return 所有的子节点的名称

* @throws KeeperException

* @throws InterruptedException

*/

public List getChildren(String path, boolean watch) throws KeeperException, InterruptedException

/**

*

* @param path 节点路径

* @param watcher 监听器

* @param stat 状态信息【可以为null】

* @return 节点数据的二进制数组【可以通过new String()转换成字符串信息】

* @throws KeeperException

* @throws InterruptedException

*/

public byte[] getData(String path, Watcher watcher, Stat stat) throws KeeperException, InterruptedException

/**

*

* @param path 节点路径

* @param watch 监听器

* @param cb 回调函数

* @param ctx 上下文参数 ？【该参数不太理解，望知道的留言讲解，谢谢】

*/

public void getData(String path, boolean watch, AsyncCallback.DataCallback cb, Object ctx)

/**

*

* @param path 节点路径

* @param data 数据

* @param version 版本号【初始通常赋值为-1，每次更新会自动+1】

* @return 状态信息

* @throws KeeperException

* @throws InterruptedException

*/

public Stat setData(String path, byte[] data, int version) throws KeeperException, InterruptedException

/**

*如果Stat为null，则节点不存在

* @param path 节点路径

* @param watch 监听器

* @return 状态信息

* @throws KeeperException

* @throws InterruptedException

*/

public Stat exists(String path, boolean watch) throws KeeperException, InterruptedException

/**

* 如果要删除的节点有子节点，会报错：KeeperException$NotEmptyException: KeeperErrorCode = DirehOxESjctory not empty for

* 如果节点不存在，会报错：KeeperException$NoNodeException: KeeperErrorCode = NoNode for

* @param path 节点路径

* @param version 版本号[version = -1 : 匹配所有的版本]

* @throws InterruptedException

* @throws KeeperException

*/

public void delete(String path, int version) throws InterruptedException, KeeperException

JAVA调用

初始化

try {

ZooKeeper　zooKeeper = new ZooKeeper("172.23.34.13:2181", 15000, event -> {

if (event.getType() == Watcher.Event.EventType.None && event.getState() == Watcher.Event.KeeperState.SyncConnected) {

System.out.println("Connectted successful.");

}

});

} catch (IOException e) {

e.printStackTrace();

}

创建节点: create

@Test

public void create() throws KeeperException, InterruptedException {

//参数：1，节点路径； 2，要存储的数据； 3，节点的权限； 4，节点的类型

String nodePath = zooKeeper.create("/java/2183", "This is Java Node 2183.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

System.out.println(nodePath);

}

获取子节点： ls

public void getChildren() throws KeeperException, InterruptedException {

List children = zooKeeper.getChildren("/", true);

for (String child : children) {

System.out.println("child: "+child);

}

}

同步获取节点内容： get

@Test

public void getData() throws KeeperException, InterruptedException {

String path = "/java";

byte[] bytes = zooKeeper.getData(path, event -> {

if (event.getType() == Watcher.Event.EventType.NodeDataChanged && path.equals(event.getPath())) {

System.out.println("Date changed.");

}

}, null);

System.out.printf("The data of %s is : %s \n",path, new String(bytes));

}

异步获取节点内容： get

@Test

public void getDataAsync() {

String path = "/java";

zooKeeper.getData(path, false, new AsyncCallback.DataCallback() {

@Override

public void processResult(int i, String s, Object o, byte[] bytes, Stat stat) {

System.out.printf("The data of %s is : %s \n",path, new String(bytes));

}

},"1000");

//休眠20秒，查看响应结果

try {

Thread.sleep(20000);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

指定版本号更新数据：set

@Test

public void setData() throws KeeperException, InterruptedException {

Stat stat = zooKeeper.setData("/java", "This is from java.".getBytes(), -1);

//更新节点后，version会自动+1。故，返回值为0

System.out.println(stat.getAversion());

http://}

多线程下更新数据：set

@Test

public void setDataThread() throws KeeperException, InterruptedException {

String path = "/java";

Stat stat = new Stat();

//1,先获取节点的当前版本

zooKeeper.getData(path,false,stat);

//2，在当前版本的基础上修改节点内容

zooKeeper.setData(path, "This is from java.".getBytes(), stat.getVersion());

}

判断节点是否存在

@Test

public hOxESjvoid exists() throws KeeperException, InterruptedException {

Stat stat = zooKeeper.exists("/java", false);

if (stat == null) {

System.out.println("Not Exists.");

}else {

System.out.println("Exists.");

}

}

删除节点

@Test

public void delete() throws KeeperException, InterruptedException {

//version = -1 : 匹配所有的版本

zooKeeper.delete("/java/2182", -1);

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。