Zookeeper使用--Java API(转)
Posted 军子
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Zookeeper使用--Java API(转)相关的知识,希望对你有一定的参考价值。
1 创建节点
创建节点有异步和同步两种方式。无论是异步或者同步,Zookeeper都不支持递归调用,即无法在父节点不存在的情况下创建一个子节点,如在/zk-ephemeral节点不存在的情况下创建/zk-ephemeral/ch1节点;并且如果一个节点已经存在,那么创建同名节点时,会抛出NodeExistsException异常。
① 同步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooKeeper; public class Zookeeper_Create_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); public static void main(String[] args) throws Exception { ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_Sync_Usage()); System.out.println(zookeeper.getState()); connectedSemaphore.await(); String path1 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("Success create znode: " + path1); String path2 = zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println("Success create znode: " + path2); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { connectedSemaphore.countDown(); } } }
运行结果如下
CONNECTING Success create znode: /zk-test-ephemeral- Success create znode: /zk-test-ephemeral-0000000043
结果表明已经成功创建了临时节点和临时顺序节点,在创建顺序节点时,系统会在后面自动增加一串数字。
② 异步方式
使用异步方式于同步方式的区别在于节点的创建过程(包括网络通信和服务端的节点创建过程)是异步的,在同步接口调用过程中,开发者需要关注接口抛出异常的可能,但是在异步接口中,接口本身不会抛出异常,所有异常都会在回调函数中通过Result Code来体现。
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; public class Zookeeper_Create_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); public static void main(String[] args) throws Exception { ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_Create_API_ASync_Usage()); System.out.println(zookeeper.getState()); connectedSemaphore.await(); zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new IStringCallback(), "I am context. "); zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, new IStringCallback(), "I am context. "); zookeeper.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, new IStringCallback(), "I am context. "); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { connectedSemaphore.countDown(); } } } class IStringCallback implements AsyncCallback.StringCallback { public void processResult(int rc, String path, Object ctx, String name) { System.out.println("Create path result: [" + rc + ", " + path + ", " + ctx + ", real path name: " + name); } }
运行结果如下
CONNECTING Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral- Create path result: [-110, /zk-test-ephemeral-, I am context. , real path name: null Create path result: [0, /zk-test-ephemeral-, I am context. , real path name: /zk-test-ephemeral-0000000045
结果表明已经成功使用异步方式创建了相应节点。
2 删除节点
只允许删除叶子节点,即一个节点如果有子节点,那么该节点将无法直接删除,必须先删掉其所有子节点。同样也有同步和异步两种方式。
① 同步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; 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 Delete_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new Delete_API_Sync_Usage()); connectedSemaphore.await(); zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path); zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path + "/c1"); try { zk.delete(path, -1); } catch (Exception e) { System.out.println("fail to delete znode: " + path); } zk.delete(path + "/c1", -1); System.out.println("success delete znode: " + path + "/c1"); zk.delete(path, -1); System.out.println("success delete znode: " + path); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } } } }
运行结果如下
success create znode: /zk-book success create znode: /zk-book/c1 fail to delete znode: /zk-book success delete znode: /zk-book/c1 success delete znode: /zk-book
结果表明若节点有子节点,则无法将其删除,必须先删除其所有子节点。
② 异步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; public class Delete_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new Delete_API_ASync_Usage()); connectedSemaphore.await(); zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path); zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path + "/c1"); zk.delete(path, -1, new IVoidCallback(), null); zk.delete(path + "/c1", -1, new IVoidCallback(), null); zk.delete(path, -1, new IVoidCallback(), null); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } } } } class IVoidCallback implements AsyncCallback.VoidCallback { public void processResult(int rc, String path, Object ctx) { System.out.println(rc + ", " + path + ", " + ctx); } }
运行结果如下
success create znode: /zk-book success create znode: /zk-book/c1 -111, /zk-book, null 0, /zk-book/c1, null 0, /zk-book, null
结果结果表明第一次删除/zk-book的时异常,ResultCode为-111。
3 子节点获取
读取节点的子节点列表,同样可以使用同步和异步的方式进行操作。
① 同步方式
package com.hust.grid.leesf.examples; import java.util.List; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.Watcher.Event.KeeperState; import org.apache.zookeeper.ZooDefs.Ids; public class Zookeeper_GetChildren_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk = null; public static void main(String[] args) throws Exception { String path = "/zk-book-1"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_GetChildren_API_Sync_Usage()); connectedSemaphore.await(); zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path); zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path + "/c1"); List<String> childrenList = zk.getChildren(path, true); System.out.println(childrenList); zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path + "/c2"); Thread.sleep(1000); zk.create(path + "/c3", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path + "/c3"); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } else if (event.getType() == EventType.NodeChildrenChanged) { try { System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true)); } catch (Exception e) { } } } } }
运行结果如下
success create znode: /zk-book-1 success create znode: /zk-book-1/c1 [c1] success create znode: /zk-book-1/c2 ReGet Child:[c1, c2] success create znode: /zk-book-1/c3 ReGet Child:[c3, c1, c2]
值得注意的是,Watcher通知是一次性的,即一旦触发一次通知后,该Watcher就失效了,因此客户端需要反复注册Watcher,即程序中在process里面又注册了Watcher,否则,将无法获取c3节点的创建而导致子节点变化的事件。
② 异步方式
package com.hust.grid.leesf.examples; import java.util.List; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; 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 Zookeeper_GetChildren_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk = null; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new Zookeeper_GetChildren_API_ASync_Usage()); connectedSemaphore.await(); zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path); zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path + "/c1"); zk.getChildren(path, true, new IChildren2Callback(), null); zk.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path + "/c2"); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } else if (event.getType() == EventType.NodeChildrenChanged) { try { System.out.println("ReGet Child:" + zk.getChildren(event.getPath(), true)); } catch (Exception e) { } } } } } class IChildren2Callback implements AsyncCallback.Children2Callback { public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) { System.out.println("Get Children znode result: [response code: " + rc + ", param path: " + path + ", ctx: " + ctx + ", children list: " + children + ", stat: " + stat); } }
运行结果如下
success create znode: /zk-book success create znode: /zk-book/c1 Get Children znode result: [response code: 0, param path: /zk-book, ctx: null, children list: [c1], stat: 2901,2901,1478226062843,1478226062843,0,1,0,0,0,1,2902 success create znode: /zk-book/c2 ReGet Child:[c1, c2]
结果表示通过异步的方式可以获取子节点信息。
4 数据节点获取
对于节点的数据获取,同样存在同步和异步两种方式。
① 同步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; 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 GetData_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk = null; private static Stat stat = new Stat(); public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new GetData_API_Sync_Usage()); connectedSemaphore.await(); zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path); System.out.println("the data of znode " + path + " is : " + new String(zk.getData(path, true, stat))); System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion()); zk.setData(path, "123".getBytes(), -1); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } else if (event.getType() == EventType.NodeDataChanged) { try { System.out.println("the data of znode " + event.getPath() + " is : " + new String(zk.getData(event.getPath(), true, stat))); System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion()); } catch (Exception e) { } } } } }
运行结果如下
success create znode: /zk-book the data of /zk-book is : 123 czxID: 2924, mzxID: 2924, version: 0 the data of /zk-book is : 123 czxID: 2924, mzxID: 2925, version: 1
结果表明可以使用getData函数获取节点的数据。
② 异步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; 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 GetData_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new GetData_API_ASync_Usage()); connectedSemaphore.await(); zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path); zk.getData(path, true, new IDataCallback(), null); zk.setData(path, "123".getBytes(), -1); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } else if (event.getType() == EventType.NodeDataChanged) { try { zk.getData(event.getPath(), true, new IDataCallback(), null); } catch (Exception e) { } } } } } class IDataCallback implements AsyncCallback.DataCallback { public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) { System.out.println("rc: " + rc + ", path: " + path + ", data: " + new String(data)); System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion()); } }
运行结果如下
success create znode: /zk-book rc: 0, path: /zk-book, data: 123 czxID: 2932, mzxID: 2932, version: 0 rc: 0, path: /zk-book, data: 123 czxID: 2932, mzxID: 2933, version: 1
结果表明采用异步方式同样可方便获取节点的数据。
5 更新数据
在更新数据时,setData方法存在一个version参数,其用于指定节点的数据版本,表明本次更新操作是针对指定的数据版本进行的,但是,在getData方法中,并没有提供根据指定数据版本来获取数据的接口,那么,这里为何要指定数据更新版本呢,这里方便理解,可以等效于CAS(compare and swap),对于值V,每次更新之前都会比较其值是否是预期值A,只有符合预期,才会将V原子化地更新到新值B。Zookeeper的setData接口中的version参数可以对应预期值,表明是针对哪个数据版本进行更新,假如一个客户端试图进行更新操作,它会携带上次获取到的version值进行更新,而如果这段时间内,Zookeeper服务器上该节点的数据已经被其他客户端更新,那么其数据版本也会相应更新,而客户端携带的version将无法匹配,无法更新成功,因此可以有效地避免分布式更新的并发问题。
① 同步方式
package com.hust.grid.leesf.examples; 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.Ids; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; public class SetData_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new SetData_API_Sync_Usage()); connectedSemaphore.await(); zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path); zk.getData(path, true, null); Stat stat = zk.setData(path, "456".getBytes(), -1); System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion()); Stat stat2 = zk.setData(path, "456".getBytes(), stat.getVersion()); System.out.println("czxID: " + stat2.getCzxid() + ", mzxID: " + stat2.getMzxid() + ", version: " + stat2.getVersion()); try { zk.setData(path, "456".getBytes(), stat.getVersion()); } catch (KeeperException e) { System.out.println("Error: " + e.code() + "," + e.getMessage()); } Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } } } }
运行结果如下
success create znode: /zk-book czxID: 2936, mzxID: 2937, version: 1 czxID: 2936, mzxID: 2938, version: 2 Error: BADVERSION,KeeperErrorCode = BadVersion for /zk-book
结果表明由于携带的数据版本不正确,而无法成功更新节点。其中,setData中的version参数设置-1含义为客户端需要基于数据的最新版本进行更新操作。
② 异步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; 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 SetData_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, new SetData_API_ASync_Usage()); connectedSemaphore.await(); zk.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); System.out.println("success create znode: " + path); zk.setData(path, "456".getBytes(), -1, new IStatCallback(), null); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } } } } class IStatCallback implements AsyncCallback.StatCallback { public void processResult(int rc, String path, Object ctx, Stat stat) { System.out.println("rc: " + rc + ", path: " + path + ", stat: " + stat); } }
运行结果如下
success create znode: /zk-book rc: 0, path: /zk-book, stat: 2942,2943,1478228414526,1478228414545,1,0,0,96876700808708136,3,0,2942
rc(ResultCode)为0,表明成功更新节点数据。
6 检测节点是否存在
在调用接口时注册Watcher的话,还可以对节点是否存在进行监听,一旦节点被创建、被删除、数据更新,都会通知客户端。
① 同步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; 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 Exist_API_Sync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, // new Exist_API_Sync_Usage()); connectedSemaphore.await(); zk.exists(path, true); zk.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); zk.setData(path, "123".getBytes(), -1); zk.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("success create znode: " + path + "/c1"); zk.delete(path + "/c1", -1); zk.delete(path, -1); Thread.sleep(Integer.MAX_VALUE); } public void process(WatchedEvent event) { try { if (KeeperState.SyncConnected == event.getState()) { if (EventType.None == event.getType() && null == event.getPath()) { connectedSemaphore.countDown(); } else if (EventType.NodeCreated == event.getType()) { System.out.println("success create znode: " + event.getPath()); zk.exists(event.getPath(), true); } else if (EventType.NodeDeleted == event.getType()) { System.out.println("success delete znode: " + event.getPath()); zk.exists(event.getPath(), true); } else if (EventType.NodeDataChanged == event.getType()) { System.out.println("data changed of znode: " + event.getPath()); zk.exists(event.getPath(), true); } } } catch (Exception e) { } } }
运行结果如下
success create znode: /zk-book data changed of znode: /zk-book success create znode: /zk-book/c1 success delete znode: /zk-book
结果表明:
· 无论节点是否存在,都可以通过exists接口注册Watcher。
· 注册的Watcher,对节点创建、删除、数据更新事件进行监听。
· 对于指定节点的子节点的各种变化,不会通知客户端。
② 异步方式
package com.hust.grid.leesf.examples; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.CreateMode; 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 Exist_API_ASync_Usage implements Watcher { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk; public static void main(String[] args) throws Exception { String path = "/zk-book"; zk = new ZooKeeper("127.0.0.1:2181", 5000, Zookeeper Api(java)入门与应用(转)Zookeeper C API 指南四(C API 概览)(转)
Exception in thread “main“ java.lang.NullPointerException at com.zs.zookeeper.DistributeServer.reg(代