005_Java操作ZooKeeper

1. ZooKeeper的JavaClient是我们更轻松的去对ZooKeeper进行各种操作。我们仅需要引入zookeeper-3.7.0.jar和zookeeper-jute-3.7.0.jar两个jar包即可。

2. 创建会话

2.1. 客户端可以创建一个ZooKeeper实例来连接ZooKeeper服务器。

2.2. ZooKeeper构造函数的connectString参数, 连接服务器列表, 使用,分割多个服务器地址。

2.3. ZooKeeper构造函数的sessionTimeout参数, 心跳检测时间周期, 以毫秒为单位。

2.4. ZooKeeper构造函数的watcher参数, 事件处理通知器。

2.5. ZooKeeper客户端和服务器会话的创建是一个异步的过程, 也就是说在程序中, ZooKeeper实例创建完成后立即返回, 但是大多数情况下我们并没有真正构建好一个可用会话, 在会话的生命周期处于CONNECTING时才算真正建立完毕。

3. 创建节点

3.1. ZooKeeper提供了两套创建节点的方法, 同步和异步创建节点方式。

3.2. 创建节点path参数, 节点路径, 不允许递归创建节点。

3.3. 创建节点data参数, 节点内容, 字节数组类型。

3.4. 创建节点acl参数, 节点权限, 最简单的值Ids.OPEN_ACL_UNSAFE开放权限。

3.5. 创建节点createMode参数, 创建节点类型。CreateMode.PERSISTENT持久节点; CreateMode.PERSISTENT_SEQUENTIAL持久顺序节点; CreateMode.EPHEMERAL临时节点; CreateMode.EPHEMERAL_SEQUENTIAL临时顺序节点。

3.6. 创建节点Create2Callback参数, 注册一个异步回调函数。rc: 服务器相应码, 0表示创建成功, -4表示端口未连接, -110表示指定节点存在, -112表示会话过期。path: 接口调用时传入API的数据节点的路径参数。xtx: 为接口传入API的上下文参数值。name: 实际在服务器端创建节点的名称。stat: 节点属性信息。

3.7. 创建节点ctx参数, 为接口调用传入上下文参数。

4. 创建节点例子

4.1. 新建一个名为zookeeper的Java项目, 同时拷入相关jar包

4.2. 创建Create.java

package com.zk.rm;import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback.Create2Callback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;public class Create {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);public static void main(String[] args) {try {// 创建连接ZooKeeper zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {// 刚刚连接成功, 没有事件if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");} }}});// 进行阻塞cdl.await();// 创建父节点String ret = zk.create("/myRoot", "创建了一个我的根节点。".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);System.out.println(System.currentTimeMillis() + " " + ret);// 异步创建子节点zk.create("/myRoot/child1", "根节点上创建了一个child1的子节点".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL, new Create2Callback() {@Overridepublic void processResult(int rc, String path, Object ctx, String name, Stat stat) {System.out.println("rc = " + rc + ", path = " + path + ", ctx = " + ctx + ", name = " + name);}}, "/myRoot/child1下文参数");zk.create("/myRoot/child2", "根节点上创建了一个child2的子节点".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new Create2Callback() {@Overridepublic void processResult(int rc, String path, Object ctx, String name, Stat stat) {System.out.println("rc = " + rc + ", path = " + path + ", ctx = " + ctx + ", name = " + name);}}, "/myRoot/child2下文参数");// 关闭连接zk.close();} catch (InterruptedException | IOException | KeeperException e) {e.printStackTrace();}}
}

4.3. 运行结果

5. 删除节点

5.1. delete删除节点也有同步删除和异步删除两套API。

5.2. 删除节点path参数, 节点名称。

5.3. 删除节点version参数, 版本号, -1跳过版本检查。

5.4. 删除节点cb参数, 一个异步回调函数。

5.5. 删除节点ctx参数, 用于传递上下文信息的对象。

5.6. 在ZooKeeper中, 只允许删除叶子节点信息, 也就是说如果节点下还有子孙节点, 必须删除所有子孙节点才能删除该节点。

6. 检查节点是否存在

6.1. exists检查节点是否存在方法也有同步检查和异步检查两套API。

6.2. 检查节点是否存在path参数, 节点名称。

6.3. 检查节点是否存在watch参数, 注册一个监听对象。

6.4. 检查节点是否存在cb参数, 一个异步回调函数。

6.5. 检查节点是否存在ctx参数, 用于传递上下文信息的对象。

7. 删除和检查节点是否存在例子

7.1. 创建ExistsDelete.java

package com.zk.rm;import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;public class ExistsDelete {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);public static void main(String[] args) {try {ZooKeeper zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");}}}});// 进行阻塞cdl.await();if(zk.exists("/myRoot", null) != null) {// 同步删除方法// zk.delete("/myRoot", -1);// 异步删除方法zk.delete("/myRoot", -1, new VoidCallback() {@Overridepublic void processResult(int rc, String path, Object ctx) {System.out.println("rc = " + rc + ", path = " + path + ", ctx = " + ctx);}}, "delete /myRoot上下文参数");}zk.close();} catch (InterruptedException | IOException | KeeperException e) {e.printStackTrace();}}
}

7.2. 运行结果

8. 获取子节点例子

8.1. 创建GetChildren.java

package com.zk.rm;import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;public class GetChildren {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);public static void main(String[] args) {try {ZooKeeper zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {// 刚刚连接成功, 没有事件if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");}}}});// 进行阻塞cdl.await();// 获取子节点List<String> children = zk.getChildren("/", false);for (String child : children) {System.out.println(child);}zk.close();} catch (InterruptedException | IOException | KeeperException e) {e.printStackTrace();}}
}

8.2. 运行结果

9. 设置和获取节点例子

9.1. 创建GetSetData.java

package com.zk.rm;import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;public class GetSetData {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);public static void main(String[] args) {try {ZooKeeper zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");}}}});// 进行阻塞cdl.await();// 获取节点byte[] data = zk.getData("/myRoot", false, null);System.out.println("根节点数据: " + new String(data));Stat stat = zk.setData("/myRoot", "modify data".getBytes(), -1);System.out.println("czxid: " + stat.getCzxid() + ", ctime: " + stat.getCtime() + ", cversion: " + stat.getCversion());System.out.println("mzxid: " + stat.getMzxid() + ", mtime: " + stat.getMtime() + ", pzxid: " + stat.getPzxid());System.out.println("version: " + stat.getVersion() + ", dataLength: " + stat.getDataLength() + ", aversion: " + stat.getAversion());System.out.println("numChildren: "+ stat.getNumChildren() + ", ephemeralOwner: " + stat.getEphemeralOwner());zk.close();} catch (InterruptedException | IOException | KeeperException e) {e.printStackTrace();}}
}

9.2. 运行结果

10. watch事件

10.1. ZooKeeper的watch事件监听节点的变化, 当节点发生变化时, 通知注册了监听的客户端。

10.2. 事件类型: EventType.NodeCreated创建节点; EventType.NodeDataChanged节点数据发生变化; EventType.NodeChildrenChanged子节点发生变化; EventType.NodeDeleted删除节点。

10.3. 状态类型: KeeperState.SyncConnected建立连接成功; KeeperState.Disconnected未连接; KeeperState.AuthFailed认证失败; KeeperState.Expired连接过期。

11. watch事件例子

11.1. 创建Watch.java

package com.zk.rm;import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.AddWatchMode;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;public class Watch {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);private static ZooKeeper zk;public static void main(String[] args) {try {zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {try {if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");} else if (EventType.NodeDataChanged == event.getType()) {System.out.println("data = " + new String(zk.getData("/myRoot", false, null)));} else if (EventType.NodeChildrenChanged == event.getType()) {System.out.println("allChildrenNumber = "+ zk.getAllChildrenNumber("/myRoot"));} else if (EventType.NodeCreated == event.getType()) {System.out.println("创建节点");} else if (EventType.NodeDeleted == event.getType()) {System.out.println("删除节点");}} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}}});// 进行阻塞cdl.await();// 添加节点监听zk.addWatch("/myRoot", AddWatchMode.PERSISTENT_RECURSIVE);String childRet2 = zk.create("/myRoot/child2", "根节点上创建了一个child2的子节点".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);System.out.println(childRet2);zk.setData("/myRoot", "修改/myRoot节点数据".getBytes(), -1);zk.delete("/myRoot/child2", -1);Thread.sleep(60000);zk.close();} catch (KeeperException | IOException | InterruptedException e) {e.printStackTrace();}}
}

11.2. 运行结果

12. ACL例子

12.1. 创建AccessControlList.java

package com.zk.rm;import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Perms;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;public class AccessControlList {private static final String connectString = "192.168.25.133:2181,192.168.25.135:2181,192.168.25.138:2181";private static final CountDownLatch cdl = new CountDownLatch(1);public static void main(String[] args) {try {ZooKeeper zk = new ZooKeeper(connectString, 5000, new Watcher() {@Overridepublic void process(WatchedEvent event) {if(KeeperState.SyncConnected == event.getState()) {if(EventType.None == event.getType()) {// 如果建立连接成功, 发送信号量, 让后续阻塞程序能够继续向下执行cdl.countDown();System.out.println("建立连接成功了。");}}}});// 进行阻塞cdl.await();List<ACL> acls = new ArrayList<ACL>();Id id1 = new Id("digest", DigestAuthenticationProvider.generateDigest("zs:123456"));ACL acl = new ACL(Perms.ALL, id1);acls.add(acl);String rt = zk.create("/myRoot", "my root data".getBytes(), acls, CreateMode.PERSISTENT);System.out.println(rt);zk.addAuthInfo("digest", "zs:123456".getBytes());System.out.println(new String(zk.getData("/myRoot", null, null)));zk.close();} catch (KeeperException | IOException | InterruptedException | NoSuchAlgorithmException e) {e.printStackTrace();}}
}

12.2. 运行结果

005_Java操作ZooKeeper相关推荐

ZooKeeper入门（二）Java操作zookeeper
首先是pom依赖: <dependency><groupId>org.apache.zookeeper</groupId><artifactId>zoo ...
zookeeper教程,docker 安装,命令,python操作zookeeper，分布式队列，分布式锁
docker安装zookeeper服务端首先安装单节点的服务端,如果安装多节点的服务端,需要为每个节点配置其他节点的地址. docker run --privileged=true -d --nam ...
Zookeeper学习之源生API的使用(java与shell操作zookeeper)。
如果不会搭建zookeeper环境:请看此文章:Zookeeper学习之集群环境搭建 1.操作zookeeper(shell) 启动zookeeper客户端:zkCli.sh; 根据提示命令进行操作: ...
使用Kazoo操作ZooKeeper服务治理
使用Kazoo操作ZooKeeper服务治理单机服务的可靠性及可扩展性有限,某台服务宕机可能会影响整个系统的正常使用:分布式服务能够有效地解决这一问题,但同时分布式服务也会带来一些新的问题,如:服务 ...
python系列之：kazoo连接Zookeeper操作Zookeeper
python系列之:kazoo连接Zookeeper操作Zookeeper 一.连接zookeeper 二.读取zookeeper节点三.读取zookeeper内容四.kazoo连接Zookeep ...
python 操作 zookeeper 快速入门
python 操作 zookeeper 快速入门文章目录 python 操作 zookeeper 快速入门什么是zookeeper python 操作 zk 快速入门 zk节点创建节点查询节点 ...
支持断线重连、永久watcher、递归操作 ZooKeeper 客户端
项目介绍 ZooKeeper本质上是一个分布式的小文件存储系统.原本是Apache Hadoop的一个组件,现在被拆分为一个Hadoop的独立子项目. Zookeeper 作为一个分布式的服务框架,主 ...
Curator操作ZooKeeper
Curator极大简化了ZooKeeper的使用,增加了针对ZooKeeper集群中connection的管理. 节点的创建和删除 import java.util.List; import java ...
ZooKeeper(二) idea中使用Java操作zookeeper
本章介绍使在idea环境中使用maven工程进行连接zookeeper,并进行简单的增删查改操作. 一.使用的pom依赖添加zookeeper依赖,以及进行测试的junit测试依赖. <?xm ...

005_Java操作ZooKeeper

005_Java操作ZooKeeper相关推荐

最新文章

热门文章