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();
        }
    }
    //創(chuàng)建子節(jié)點
    private String createSaNode() throws KeeperException, InterruptedException {
        // 如果根節(jié)點不存在，則創(chuàng)建根節(jié)點
        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");
        // 創(chuàng)建EPHEMERAL_SEQUENTIAL類型節(jié)點
        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<String> prevSaIds;
     public DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
                       DataMonitorListener listener) {
        this.zk = zk;
        this.znode = znode;
        this.chainedWatcher = chainedWatcher;
        this.listener = listener;
        // 這是整個監(jiān)控的真正開始，通過獲取children節(jié)點開始。設置了本對象為監(jiān)控對象，回調對象也是本對象。以后均是事件驅動。
        zk.getChildren(znode, true, this, null);
    }
 
    /**
     * 其他和monitor產生交互的類，需要實現此listener
     */
    public interface DataMonitorListener {
        /**
         * The existence status of the node has changed.
         */
        void changed(List<String> saIds);
 
        /**
         * The ZooKeeper session is no longer valid.
         *
         * @param rc
         *                the ZooKeeper reason code
         */
        void closing(int rc);
    }
 
   /*
    *監(jiān)控/saids的回調函數。除了處理異常外。
    *如果發(fā)生變化，和構造函數中一樣，通過getChildren，再次監(jiān)控，并處理children節(jié)點變化后的業(yè)務
    */
    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節(jié)點后的回調函數。
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> 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<String> saIds = null;
 
        //如果存在，再次查詢到最新children，此時僅查詢，不要設置監(jiān)控了
        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的監(jiān)控者，監(jiān)控和zookeeper連接的變化
         * 自己作為DataMonitor的listener。當dm監(jiān)控到變化時會調用executor執(zhí)行業(yè)務操作
         */
    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監(jiān)控者的回調，直接傳遞事件給monitor的回調函數統(tǒng)一處理
     */
    @Override
    public void process(WatchedEvent event) {
        dm.process(event);
    } 
    /*
     *當關閉時，讓線程線繼續(xù)走完
     */
    public void closing(int rc) {
        synchronized (this) {
            notifyAll();
        }
    }
 
    /*
     *監(jiān)控到/saids變化后的處理類
     */
    static class SaIdsLoader extends Thread {
         List<String> saIds = null; 
        //構造對象后直接啟動線程
        public SaIdsLoader(List<String> saIds){
            this.saIds = saIds;
            start();
        }
 
        public void run() {
            System.out.println("------------加載開始------------");
            //業(yè)務處理的地方
            if(saIds!=null){
                saIds.forEach(id->{
                    System.out.println(id);
                });
            }
            System.out.println("------------加載結束------------");
        }
    }
 
    /*
     *作為listener對外暴露的方法，在節(jié)點/saids變化時被調用。
     */
    @Override
    public void changed(List<String> data) {
                new SaIdsLoader(data);
    }
}