通过创建一个zookeeper实例来链接zookeeper服务器。
注意:zookeeper客户端和服务器端会话的建立是一个异步的过程,也就是说在程序中,我们程序方法在处理完客户端初始化后,立即返回(程序往下执行代码,这样,大多数情况下我们并没有真正构建好一个可用会话,在会话的声明周期处于"CONNECTING"时才算真正建立完毕,所以我们需要使用多线程中的一个工具类CountDownLatch)
构造函数如下所示:
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher,
boolean canBeReadOnly, HostProvider aHostProvider, ZKClientConfig clientConfig)
public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long sessionId,
byte[] sessionPasswd, boolean canBeReadOnly, HostProvider aHostProvider,
ZKClientConfig clientConfig)
监听一个事件
public class CreateSession implements Watcher { private static ZooKeeper zookeeper; public static void main(String[] args) throws IOException, InterruptedException { zookeeper = new ZooKeeper("127.0.0.0:2181",5000,new CreateSession()); System.out.println(zookeeper.getState());//状态 Thread.sleep(Integer.MAX_VALUE); } private void doSomething(){ System.out.println("业务处理。。。"); } @Override public void process(WatchedEvent event) { System.out.println("event = "+event); if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(); } } } }
创建同步节点
public class CreateNodeSync implements Watcher { private static ZooKeeper zookeeper; public static void main(String[] args) throws IOException, InterruptedException { zookeeper = new ZooKeeper("192.168.1.105:2181",5000,new CreateNodeSync()); System.out.println(zookeeper.getState()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(){ try {
String path = zookeeper.create("/node_4", "张三".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("return path:"+path); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("业务处理。。。"); } @Override public void process(WatchedEvent event) { System.out.println("event="+event); if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(); } } } }
其中 ,zookeeper.create() 方法有四个参数:
参数1,节点路径(名称):/nodeName (不允许递归创建节点,也就是说在父节点不存在的情况下,不允许创建子节点)
参数2,节点内容:要求类型是字节数组 (不支持序列化方式,如果需要实现序列化,可使用java相关的序列化框架,如Hession,Kryo框架)
参数3,节点权限:使用Ids.OPEN_ACL_UNSAFE开发权限即可(一般在权限没有太高要求的场景下,没必要关注),也可以使用如下方法
ACL aclIp = new ACL(Perms.READ,new Id("ip","127.0.0.0"));
ACL aclDigest = new ACL(Perms.READ|Perms.WRITE,new Id("user1",DigestAuthenticationProvider.generateDigest("user1:123456")));
ArrayList<ACL> acls = new ArrayList<ACL>();
acls.add(aclDigest);
acls.add(aclIp);
参数4,节点类型 创建节点的类型:CreateMode.* 提供四种节点类型
PERSISTENT (持久节点)
PERSISTENT_SEQUENTIAL(持久顺序节点)
EPHEMRAL(临时节点)
EPHEMRAL_SEQUENTIAL(临时顺序节点)
异步创建节点
public class CreateNodeASync implements Watcher { private static ZooKeeper zookeeper; public static void main(String[] args) throws IOException, InterruptedException { zookeeper = new ZooKeeper("127.0.0.0:2181",5000,new CreateNodeASync()); System.out.println(zookeeper.getState()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(){ // 异步时,无返回值,IStringCallback 在这里是异步回调接口 zookeeper.create("/node_5", "张三".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT,new IStringCallback(),"AAAAA"); } @Override public void process(WatchedEvent event) { System.out.println("event="+event); if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(); } } } static class IStringCallback implements AsyncCallback.StringCallback{ @Override public void processResult(int rc, String path, Object ctx, String name) {
//rc:为服务端响应码, 0表示调用成功,-4表示端口连接,-110表示指定节点存在,-112表示会话已经过期
//path:接口调用时传入API的数据节点的路径参数
//ctx:为调用接口传入API的ctx的值
//name:实际在服务器端创建节点的名称,有序时,会在节点名字后面加类似 000001
} } }
检测节点是都存在
exists方法
参数1 path:路径
参数2 watcher :注册的watcher对象。一旦之后节点内容有变更,则会向客户端发送通知,该参数允许为null (用于三类事件监听:节点的创建,删除,更新)
参数3 watch :是否使用watcher,如果true 则使用默认上文中的watcher,false则不使用watcher
参数4 cb:回调函数
参数5 ctx:用于传递的下文信息对象
注意:exists 方法意义在与无论节点是都存在,都可以进行注册watcher,能够对节点的创建,删除和修改进行监听,但是其子节点发生各种变化,都不会通知客户端。
同步判断节点是否存在
public class NodeExistsSync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new NodeExistsSync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zooKeeper){ try{ Stat stat = zooKeeper.exists("/node_1", true); System.out.println(stat); }catch(Exception e){ e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ try { if (event.getType()==EventType.NodeCreated){ System.out.println(event.getPath()+" created"); System.out.println(zooKeeper.exists(event.getPath(), true)); } else if (event.getType()==EventType.NodeDataChanged){ System.out.println(event.getPath()+" updated"); System.out.println(zooKeeper.exists(event.getPath(), true)); } else if (event.getType()==EventType.NodeDeleted){ System.out.println(event.getPath()+" deleted"); System.out.println(zooKeeper.exists(event.getPath(), true)); } } catch (Exception e) { e.printStackTrace(); } } } } }
异步判断节点是否存在
public class NodeExistsASync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new NodeExistsASync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zookeeper){ zooKeeper.exists("/node_1", true, new IStateCallback(), null); } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ try { if (event.getType()==EventType.NodeCreated){ System.out.println(event.getPath()+" created"); zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); } else if (event.getType()==EventType.NodeDataChanged){ System.out.println(event.getPath()+" updated"); zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); } else if (event.getType()==EventType.NodeDeleted){ System.out.println(event.getPath()+" deleted"); zooKeeper.exists(event.getPath(), true, new IStateCallback(), null); } } catch (Exception e) { e.printStackTrace(); } } } } static class IStateCallback implements AsyncCallback.StatCallback{ @Override public void processResult(int rc, String path, Object ctx, Stat stat) { System.out.println("rc:"+rc); } } }
删除节点:
delete方法(api提供了两个接口,同步删除和异步删除方式)
同步方式:
参数1,节点名称 /deletePath
参数2,版本号,即表示本次删除操作是针对该数据的某个版本进行操作。
异步方式:
参数3:一个异步回调函数
参数4:用于传递上下文信息的对象。
注意:在zookeeper中,只允许删除叶子节点信息,也就是说如果当前节点不是叶子节点则无法删除,或必须先删除其下所有的子节点,
同步删除节点
public class DeleteNodeSync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new DeleteNodeSync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zooKeeper){ try { zooKeeper.delete("/node_7", -1); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); } } } }
异步删除节点
public class DeleteNodeASync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new DeleteNodeASync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(WatchedEvent event){ zooKeeper.delete("/node_6", -1, new IVoidCallback(),null); } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(event); } } } static class IVoidCallback implements AsyncCallback.VoidCallback{ @Override public void processResult(int rc, String path, Object ctx) { StringBuilder sb = new StringBuilder(); sb.append("rc="+rc).append(" "); sb.append("path"+path).append(" "); sb.append("ctx="+ctx).append(" "); System.out.println(sb.toString()); } } }
同步修改节点
public class UpdateNodeSync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new UpdateNodeSync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zooKeeper){ try { Stat stat = zooKeeper.setData("/node_6", "123".getBytes(), -1); System.out.println("stat:"+stat); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); } } } }
异步修改节点
public class UpdateNodeASync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new UpdateNodeASync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(WatchedEvent event){ zooKeeper.setData("/node_6", "234".getBytes(), -1, new IStatCallback(),null); } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(event); } } } static class IStatCallback implements AsyncCallback.StatCallback{ @Override public void processResult(int rc, String path, Object ctx, Stat stat) { StringBuilder sb = new StringBuilder(); sb.append("rc="+rc).append(" "); sb.append("path"+path).append(" "); sb.append("ctx="+ctx).append(" "); sb.append("Stat="+stat).append(" "); System.out.println(sb.toString()); } } }
同步获取权限
public class GetDataSyncAuth implements Watcher{ private static ZooKeeper zooKeeper; private static Stat stat = new Stat(); public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataSyncAuth()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zookeeper){ zooKeeper.addAuthInfo("user2", "user2:1234".getBytes()); try { System.out.println(new String(zooKeeper.getData("/node_4", true, stat))); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ if (event.getType()==EventType.NodeDataChanged){ try { System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat))); System.out.println("stat:"+stat); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } }
同步获取节点数据
public class GetDataSync implements Watcher{ private static ZooKeeper zooKeeper; private static Stat stat = new Stat(); public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataSync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zookeeper){ zookeeper.addAuthInfo("user2", "user2:123456".getBytes()); try { System.out.println(new String(zooKeeper.getData("/node_4", true, stat))); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ if (event.getType()==EventType.NodeDataChanged){ try { System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat))); System.out.println("stat:"+stat); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } }
异步获取节点数据
public class GetDataASync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException,KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetDataASync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zookeeper){ zooKeeper.getData("/node_1", true, new IDataCallback(), null); } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ if (event.getType()==EventType.NodeDataChanged){ try { zooKeeper.getData(event.getPath(), true, new IDataCallback(), null); } catch (Exception e) { e.printStackTrace(); } } } } } static class IDataCallback implements AsyncCallback.DataCallback{ @Override public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) { try { System.out.println(new String(zooKeeper.getData(path, true, stat))); System.out.println("stat:"+stat); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } } }
获取子节点列表和数据
getChildren读取数据方法:包括子节点列表获取和子节点数据的获取。
参数1 path:获取指定节点下的数据(获取子节点列表)
参数2 watcher: 注册的watcher ,一旦在本次子节点获取后,子节点列表发生变更的话,那么就会向客户端发送通知,该参数允许为null
参数3 wath:表明是否需要注册一个watcher:如果为true,则会使用到zookeeper客户端上下文中提到的那个默认watcher,如果false,则表明不需要注册Watcher,
参数4 cb: 回调函数
参数5 ctx:上下文信息
参数6 stat :指定数据节点的节点状态信息
注意:当我们获取指定节点的子节点列表后,还需要订阅这个子节点列表的变化通知,这时候就可以通过注册一个watcher来实现,当子节点被添加或删除时,服务器端就会触发一个“NodeChildrenChanged”类型的时间通知,需要注意的是服务端发送给客户端的事件通知中,是不包含最新的节点列表的,客户端必须主动重新进行获取,通常在客户端收到这个事件通知后,就可以再次主动获取最新的子节点列表,也就是说,zookeeper服务端在想客户端发送watcher“NodeChildrenChanged”事件通知的时候,仅仅只发了一个通知,不会把节点的变化情况发给客户端,需要客户端自己重新获取,另外Watcher通知是一次性的,即触发后失效,因此客户端需要反复注册Watcher才行。
同步获取子节点数据
public class GetChildrenSync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("127.0.0.0:2181",5000,new GetChildrenSync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zooKeeper){ try { List<String> children = zooKeeper.getChildren("/", true); System.out.println(children); } catch (Exception e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ if (event.getType()==EventType.NodeChildrenChanged){ try { System.out.println(zooKeeper.getChildren(event.getPath(), true)); } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } } } } }
异步获取子节点数据
public class GetChildrenASync implements Watcher{ private static ZooKeeper zooKeeper; public static void main(String[] args) throws IOException, InterruptedException, KeeperException { zooKeeper = new ZooKeeper("192.168.1.105:2181",5000,new GetChildrenASync()); System.out.println(zooKeeper.getState().toString()); Thread.sleep(Integer.MAX_VALUE); } private void doSomething(ZooKeeper zookeeper){ try { zooKeeper.getChildren("/", true, new IChildren2Callback(), null); } catch (Exception e) { e.printStackTrace(); } } @Override public void process(WatchedEvent event) { if (event.getState()==KeeperState.SyncConnected){ if (event.getType()==EventType.None && null==event.getPath()){ doSomething(zooKeeper); }else{ if (event.getType()==EventType.NodeChildrenChanged){ zooKeeper.getChildren(event.getPath(), true, new IChildren2Callback(), null); } } } } static class IChildren2Callback implements AsyncCallback.Children2Callback{ @Override public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) { StringBuilder sb = new StringBuilder(); sb.append("rc="+rc).append(" "); sb.append("path="+path).append(" "); sb.append("ctx="+ctx).append(" "); sb.append("children="+children).append(" "); sb.append("stat="+stat).append(" "); System.out.println(sb.toString()); } } }
package bjsxt.zookeeper.base; import java.util.concurrent.CountDownLatch; 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; /** * Zookeeper base学习笔记 */ public class ZookeeperBase { /** zookeeper地址 */ static final String CONNECT_ADDR = "192.168.2.2:2181";// 集群可以写多个,英文逗号隔开 /** session超时时间 */ static final int SESSION_OUTTIME = 2000;// ms /** 信号量,阻塞程序执行,用于等待zookeeper连接成功,发送成功信号 */ static final CountDownLatch connectedSemaphore = new CountDownLatch(1); public static void main(String[] args) throws Exception { ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher() { @Override public void process(WatchedEvent event) { // 获取事件的状态 KeeperState keeperState = event.getState(); EventType eventType = event.getType(); // 如果是建立连接 if (KeeperState.SyncConnected == keeperState) { if (EventType.None == eventType) { // 如果建立连接成功,则发送信号量,让后续阻塞程序向下执行 System.out.println("zk 建立连接"); connectedSemaphore.countDown(); } } } }); // 进行阻塞 connectedSemaphore.await(); System.out.println(".."); // 创建父节点 // zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, // CreateMode.PERSISTENT); // 创建子节点 // zk.create("/testRoot/children", "children data".getBytes(), // Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); // 获取节点洗信息 // byte[] data = zk.getData("/testRoot", false, null); // System.out.println(new String(data)); // System.out.println(zk.getChildren("/testRoot", false)); // 修改节点的值 // zk.setData("/testRoot", "modify data root".getBytes(), -1); // byte[] data = zk.getData("/testRoot", false, null); // System.out.println(new String(data)); // 判断节点是否存在 // System.out.println(zk.exists("/testRoot/children", false)); // 删除节点 // zk.delete("/testRoot/children", -1); // System.out.println(zk.exists("/testRoot/children", false)); zk.close(); } }