ZooKeeper开发实际应用案例实战

 public class ServiceRegister implements Runnable{ private ZooKeeper zk; private static final String ZNODE = "/sas"; private static final String SA_NODE_PREFIX = "sa_"; private String hostName="localhost:2181"; public void setHostName(String hostName) { this.hostName = hostName; } public ServiceRegister() throws IOException { zk = new ZooKeeper(hostName, 10000,null); } @Override public void run() { try { createSaNode(); synchronized (this) { wait(); } } catch (KeeperException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } } //测试用 public static void main(String[] args){ try { new ServiceRegister().run(); } catch (IOException e) { e.printStackTrace(); } } //创建子节点 private String createSaNode() throws KeeperException, InterruptedException { // 如果根节点不存在，则创建根节点 Stat stat = zk.exists(ZNODE, false); if (stat == null) { zk.create(ZNODE, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } String hostName = System.getenv("HOSTNAME"); // 创建EPHEMERAL_SEQUENTIAL类型节点 String saPath = zk.create(ZNODE + "/" + SA_NODE_PREFIX, hostName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); return saPath; } }

 public class DataMonitor implements Watcher, AsyncCallback.ChildrenCallback { ZooKeeper zk; String znode; Watcher chainedWatcher; boolean dead; DataMonitorListener listener; List prevSaIds; public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher, DataMonitorListener listener) { this.zk = zk; this.znode = znode; this.chainedWatcher = chainedWatcher; this.listener = listener; // 这是整个监控的真正开始，通过获取children节点开始。设置了本对象为监控对象，回调对象也是本对象。以后均是事件驱动。 zk.getChildren(znode, true, this, null); } /** * 其他和monitor产生交互的类，需要实现此listener */ public interface DataMonitorListener { /** * The existence status of the node has changed. */ void changed(List saIds); /** * The ZooKeeper session is no longer valid. * * @param rc *                the ZooKeeper reason code */ void closing(int rc); } /* *监控/saids的回调函数。除了处理异常外。 *如果发生变化，和构造函数中一样，通过getChildren，再次监控，并处理children节点变化后的业务 */ public void process(WatchedEvent event) { String path = event.getPath(); if (event.getType() == Event.EventType.None) { // We are are being told that the state of the // connection has changed switch (event.getState()) { case SyncConnected: // In this particular example we don't need to do anything // here - watches are automatically re-registered with // server and any watches triggered while the client was // disconnected will be delivered (in order of course) break; case Expired: // It's all over dead = true; listener.closing(Code.SESSIONEXPIRED.intValue()); break; } } else { if (path != null && path.equals(znode)) { // Something has changed on the node, let's find out zk.getChildren(znode, true, this, null); } } if (chainedWatcher != null) { chainedWatcher.process(event); } } //拿到Children节点后的回调函数。 @Override public void processResult(int rc, String path, Object ctx, List children) { boolean exists; switch (rc) { case Code.Ok: exists = true; break; case Code.NoNode: exists = false; break; case Code.SessionExpired: case Code.NoAuth: dead = true; listener.closing(rc); return; default: // Retry errors zk.getChildren(znode, true, this, null); return; } List saIds = null; //如果存在，再次查询到最新children，此时仅查询，不要设置监控了 if (exists) { try { saIds = zk.getChildren(znode,null); } catch (KeeperException e) { // We don't need to worry about recovering now. The watch // callbacks will kick off any exception handling e.printStackTrace(); } catch (InterruptedException e) { return; } } //拿到最新saids后，通过listener（executor），加载Saids。 if ((saIds == null && saIds != prevSaIds) || (saIds != null && !saIds.equals(prevSaIds))) { listener.changed(saIds); prevSaIds = saIds; } } }

 public class LoadSaIdsExecutor implements Watcher, Runnable, DataMonitor.DataMonitorListener { private DataMonitor dm; private ZooKeeper zk; private static final String znode = "/sas"; private String hostName="localhost:2181"; public void setHostName(String hostName) { this.hostName = hostName; } /* *初始化zookeeper及DataMonitor * 自己作为zookeeper的监控者，监控和zookeeper连接的变化 * 自己作为DataMonitor的listener。当dm监控到变化时会调用executor执行业务操作 */ public LoadSaIdsExecutor() throws KeeperException, IOException { zk = new ZooKeeper(hostName, 300000, this); dm = new DataMonitor(zk, znode, null, this); } /** * 入口方法,测试用。 */ public static void main(String[] args) { try { new LoadSaIdsExecutor().run(); } catch (Exception e) { e.printStackTrace(); } } /** * 作为单独线程运行 */ public void run() { try { synchronized (this) { while (!dm.dead) { wait(); } } } catch (InterruptedException e) { } } /* *作为zookeeper监控者的回调，直接传递事件给monitor的回调函数统一处理 */ @Override public void process(WatchedEvent event) { dm.process(event); } /* *当关闭时，让线程线继续走完 */ public void closing(int rc) { synchronized (this) { notifyAll(); } } /* *监控到/saids变化后的处理类 */ static class SaIdsLoader extends Thread { List saIds = null; //构造对象后直接启动线程 public SaIdsLoader(List saIds){ this.saIds = saIds; start(); } public void run() { System.out.println("------------加载开始------------"); //业务处理的地方 if(saIds!=null){ saIds.forEach(id->{ System.out.println(id); }); } System.out.println("------------加载结束------------"); } } /* *作为listener对外暴露的方法，在节点/saids变化时被调用。 */ @Override public void changed(List data) { new SaIdsLoader(data); } }