zookeeper-watcher- 简单例子

zookeeper watcher示例

示例说明：zookeeper server端，client端每100毫秒写入数据，另一个client端watch节点数据变化。

server端
参考这篇的server端。
http://xkorey.iteye.com/blog/2201301

数据生成端

public void increase(){
        int inx=0;
        try {
            zk = new ZooKeeper("localhost:22801",2000,new ZKWatcher());
            if(zk.exists(path,false)==null){
                zk.create(path,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
            }
            while (run){
                zk.setData(path,(""+(inx++)).getBytes(),-1);
                Thread.sleep(100);
            }
            System.out.println("maker ext. the value is:"+inx);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }

watch端

public class ZKEventCatcher implements Watcher, AsyncCallback.StatCallback{

    String path = "/test_data_increase";
    ZooKeeper zk=null;
    Object o =new Object();

    public void init(){
        try {
            zk = new ZooKeeper("localhost:8610",3000,this);
            zk.exists(path,this);
            synchronized (o){
                o.wait();
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void processResult(int i, String s, Object o, Stat stat) {
        boolean exists;
        switch (KeeperException.Code.get(i)) {
            case OK:
                exists = true;
                break;
            case NONODE:
                exists = false;
                break;
            case SESSIONEXPIRED:
            case NOAUTH:
                return;
            default:
                zk.exists(s, true, this, stat);
                return;
        }
        if(exists){
            try {
                String val = new String(zk.getData(path,false,null));
                System.out.println(val);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void process(WatchedEvent event) {
        String path = event.getPath();
        if (event.getType() == Event.EventType.None) {
            switch (event.getState()) {
                case SyncConnected:
                    break;
                case Expired:
                    break;
            }
        } else {
            if (path != null) {
                zk.exists(path, true, this, null);
            }
        }
    }

    public static void main(String[]args){
        ZKEventCatcher catcher = new ZKEventCatcher();
        catcher.init();
    }
}

监听的结果是得到的数值是连续的、和数据生成端是一致的，并没有出现数值跳跃的情况。

结论：只要实现Watcher, AsyncCallback.StatCallback2 个类的方法。节点值变化zookeeper会触发这2个回调函数。所以只要在一开始exists的watcher设定为自己(this)。然后wait()就可以了。

官方链接：
http://zookeeper.apache.org/doc/trunk/javaExample.html