zookeeper小入门（二）

摘要：就是上面提到的变化后执行的程序。这段代码是的构造函数，除了里面的一些赋值函数，重点说下这个函数它有个参数，第一个参数是你要的名字，类似这样的。

昨晚微信开放了JS接口。忙活了一晚上。
然后周六就没啥斗志了，突然想起第二篇说好要介绍demo的没介绍，就赶紧来写了。

先来个 传送门

这里简单介绍了下官方demo。用来演示Zookeeper官方推荐的程序框架（Executor、DataMonitor）。

这个Demo实现了如下功能：

监视一个结点，获取该结点的数据，并启动一个自定义程序

如果这个结点发生变化了（数据更改，新增或者删除），则停止这个自定义程序并重新执行。

如果这个结点被删除了，则停止运行自定义程序

对着Demo看下代码。首先，官方有这么个章节

  Program Design
  
  
Conventionally, ZooKeeper applications are broken into two units, one which maintains the connection, and the other which monitors data. In this application, the class called the Executor maintains the ZooKeeper connection, and the class called the DataMonitor monitors the data in the ZooKeeper tree. Also, Executor contains the main thread and contains the execution logic. It is responsible for what little user interaction there is, as well as interaction with the exectuable program you pass in as an argument and which the sample (per the requirements) shuts down and restarts, according to the state of the znode.

通俗翻译下，官方认为，一个zookeeper应用简单地分为两个单元即可，一个用来管理连接，另一个用来监视数据。在我们的Demo中，Executor被用来管理和Zookeeper的连接，DataMonitor用来监视Zookeeper中的数据树。另外，Executor同时也包含了主线程的执行（也就是Main函数），然后也包含一点点的执行逻辑。Demo对用户交互的部分非常少，输入的几个参数（好吧，我自己去实现的时候，偷懒去写死了）就可以开始跑了，Demo还会对znode的变化或者新的znode的连入进行一些响应。

看完这段，对原文最下面的完整样例算了解一些，那这里既然就两个类，Executor和DataMonitor嘛，对吧。就分别看下。

javapublic Executor(String hostPort, String znode, String filename,
                    String exec[]) throws KeeperException, IOException {
    this.filename = filename;
    this.exec = exec;
    zk = new ZooKeeper(hostPort, 3000, this);
    dm = new DataMonitor(zk, znode, null, this);
}

javapublic void process(WatchedEvent event);

javapublic DataMonitor(ZooKeeper zk, String znode, Watcher chainedWatcher,
                       DataMonitorListener listener) {
    this.zk = zk;
    this.znode = znode;
    this.chainedWatcher = chainedWatcher;
    this.listener = listener;
    // Get things started by checking if the node exists. We are going
    // to be completely event driven
    zk.exists(znode, true, this, null);
}

java public void processResult(int rc, String path, Object ctx, Stat stat) {
        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.exists(znode, true, this, null);
                return;
        }

        byte b[] = null;
        if (exists) {
            try {
                b = zk.getData(znode, false, 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;
            }
        }
        if ((b == null && b != prevData)
                || (b != null && !Arrays.equals(prevData, b))) {
            listener.exists(b);
            prevData = b;
        }
    }

java/**
     * Return the stat of the node of the given path. Return null if no such a
     * node exists.
     * 

     * If the watch is non-null and the call is successful (no exception is thrown),
     * a watch will be left on the node with the given path. The watch will be
     * triggered by a successful operation that creates/delete the node or sets
     * the data on the node.
     *
     * @param path the node path
     * @param watcher explicit watcher
     * @return the stat of the node of the given path; return null if no such a
     *         node exists.
     * @throws KeeperException If the server signals an error
     * @throws InterruptedException If the server transaction is interrupted.
     * @throws IllegalArgumentException if an invalid path is specified
     */
    public Stat exists(final String path, Watcher watcher)
        throws KeeperException, InterruptedException;

javapublic void process(WatchedEvent event) {
    dm.process(event);
}

javapublic 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:
                logger.debug("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:
                logger.debug("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.exists(znode, true, this, null);
        }
    }
    if (chainedWatcher != null) {
        chainedWatcher.process(event);
    }
}

java public void exists(byte[] data) {
    if (data == null) {
        if (child != null) {
            System.out.println("Killing process");
            child.destroy();
            try {
                child.waitFor();
            } catch (InterruptedException e) {
            }
        }
        child = null;
    } else {
        if (child != null) {
            System.out.println("Stopping child");
            child.destroy();
            try {
                child.waitFor();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        try {
            FileOutputStream fos = new FileOutputStream(filename);
            fos.write(data);
            fos.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        try {
            System.out.println("Starting child");
            child = Runtime.getRuntime().exec(exec);
            new StreamWriter(child.getInputStream(), System.out);
            new StreamWriter(child.getErrorStream(), System.err);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}