参考：从Paxos到Zookeeper分布式一致性原理和实践
使用的zk依赖是cdh5.16.2的3.4.5
1 2 3 4 5 6 7 <!-- zookeeper --> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.5-cdh5.16.2</version> </dependency>
代码，在初始化zk的时候，会触发一个watchEvent，将CountDownLatch-1=0，从而开始读取/app1的value，此时为12345，
然后将其修改成123，会再次触发一个watchEvent，此时watchEvent的type是NodeDataChanged，之后10秒延时过后，输出/app1的value，此时为123
读取节点的getData放中的true表示是否注册一个watch，注意这里的watch是一次性的，触发通过之后就失效，需要反复注册watch
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 package com.bigdata.zookeeper; import org.apache.zookeeper.AsyncCallback; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; import java.util.concurrent.CountDownLatch; public class ZkExample implements Watcher { public static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static Stat stat = new Stat(); public static void main(String[] args) throws Exception { ZooKeeper zk = new ZooKeeper("master:2181", 5000, new ZkExample()); System.out.println(zk.getState()); try { connectedSemaphore.await(); System.out.println(new String(zk.getData("/app1", true, stat))); Thread.sleep(10000); // 10秒延时 System.out.println(new String(zk.getData("/app1", true, stat))); } catch (InterruptedException e) { System.out.println("Zk session established" + e); } } @Override public void process(WatchedEvent watchedEvent) { System.out.println(watchedEvent); if (Event.KeeperState.SyncConnected == watchedEvent.getState()) { connectedSemaphore.countDown(); } } }
输出
异步代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 package com.bigdata.zookeeper; import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import java.util.List; import java.util.concurrent.CountDownLatch; public class ZkExample implements Watcher { public static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static Stat stat = new Stat(); private static ZooKeeper zk = null; public static void main(String[] args) throws Exception { zk = new ZooKeeper("master:2181", 5000, new ZkExample()); System.out.println(zk.getState()); try { connectedSemaphore.await(); // 注册一个watch，将能异步获取到为空的子节点 zk.getChildren("/app1", true, new IChildren2Callback(), null); // 创建一个子节点 zk.create("/app1/app12", "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); Thread.sleep(10000); // 10秒延时 } catch (InterruptedException e) { System.out.println("Zk session established" + e); } } @Override public void process(WatchedEvent watchedEvent) { System.out.println(watchedEvent); if (Event.KeeperState.SyncConnected == watchedEvent.getState()) { if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) { connectedSemaphore.countDown(); } else if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) { try { System.out.println(zk.getChildren(watchedEvent.getPath(), true)); } catch (Exception e) { } } } } } class IChildren2Callback implements AsyncCallback.Children2Callback { @Override public void processResult(int i, String s, Object o, List<String> list, Stat stat) { System.out.println("Get children znode : [response code: " + i + ", path: " + s + ", children list" + list + ", stat : " + stat); } }
输出
