**************************************************************************************************************************
一个简单的手表客户端
要求
程序设计
执行者类
DataMonitor类
完成源代码列表
************************************************************************************************************************
一个简单的手表客户端
为了向您介绍ZooKeeper Java API，我们在这里开发了一个非常简单的手表客户端。这个ZooKeeper客户端监视一个ZooKeeper节点的变化，并通过启动或停止一个程序做出响应。
要求
客户有四个要求：
- 它采用以下参数：
- ZooKeeper服务的地址
- 然后是znode的名称 - 要观看的节点
- 一个带参数的可执行文件。
它获取与znode相关联的数据并启动可执行文件。
如果znode更改，客户端将重新获取内容并重新启动可执行文件。
如果znode消失，客户端会杀死可执行文件。
程序设计
通常，ZooKeeper应用程序分为两个单元，一个维护连接，另一个监视数据。在这个应用程序中，称为Executor的类维护ZooKeeper连接，而调用DataMonitor的类监视ZooKeeper树中的数据。此外，Executor包含主线程并包含执行逻辑。它负责什么小的用户交互，以及与作为参数传递的可执行程序的交互，以及示例（根据需求）关闭和重新启动，根据znode的状态。
执行者类
Executor对象是示例应用程序的主容器。 它包含ZooKeeper对象，DataMonitor，如上所述在程序设计。
// from the Executor class...
public static void main(String[] args) {
if (args.length < 4) {
System.err
.println("USAGE: Executor hostPort znode filename program [args ...]");
System.exit(2);
}
String hostPort = args[0];
String znode = args[1];
String filename = args[2];
String exec[] = new String[args.length - 3];
System.arraycopy(args, 3, exec, 0, exec.length);
try {
new Executor(hostPort, znode, filename, exec).run();
} catch (Exception e) {
e.printStackTrace();
}
}
public 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);
}
public void run() {
try {
synchronized (this) {
while (!dm.dead) {
wait();
}
}
} catch (InterruptedException e) {
}
}
回想一下，Executor的工作是启动和停止在命令行上传递其名称的可执行文件。 它响应由ZooKeeper对象发起的事件。 正如你可以在上面的代码中看到的，Executor传递一个引用自己作为ZooKeeper构造函数中的Watcher参数。 它还将对自身的引用传递给DataMonitor构造函数的DataMonitorListener参数。 根据执行者的定义，它实现这两个接口：
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...
Watcher界面由ZooKeeper Java API定义。 ZooKeeper使用它来回传到它的容器。 它只支持一个方法，process（）和ZooKeeper使用它来通信主线程将被插入的通用事件，例如ZooKeeper连接或ZooKeeper会话的状态。此示例中的执行者简单地将这些事件 到DataMonitor来决定如何处理它们。 它只是为了说明一点，按照惯例，执行器或一些类似执行器的对象“拥有”ZooKeeper连接，但它可以将事件委托给其他事件到其他对象。 它也使用此作为默认通道来触发观察事件。
public void process(WatchedEvent event) {
dm.process(event);
}
另一方面，DataMonitorListener接口不是ZooKeeper API的一部分。 它是一个完全自定义的接口，为此示例应用程序设计。 DataMonitor对象使用它来回传到它的容器，这也是Executor对象。DataMonitorListener接口如下所示：
public interface DataMonitorListener {
/**
* 节点的存在状态已更改。
*/
void exists(byte data[]);
/**
*ZooKeeper会话不再有效。
*
* @param rc
* the ZooKeeper reason code
*/
void closing(int rc);
}
此接口在DataMonitor类中定义并在Executor类中实现。 当Executor.exists（）被调用时，Executor决定是否根据需求启动或关闭。 回想一下，当znode停止存在时，需要说明杀死可执行文件。
当Executor.closing（）被调用时，Executor决定是否关闭自己以响应ZooKeeper连接永久消失。
正如你可能已经猜到的，DataMonitor是调用这些方法的对象，以响应ZooKeeper的状态的更改。
这里是Executor的DataMonitorListener.exists（）和DataMonitorListener.closing的实现：
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();
}
}
}
public void closing(int rc) {
synchronized (this) {
notifyAll();
}
}
DataMonitor类
DataMonitor类具有ZooKeeper逻辑的类。 它主要是异步和事件驱动。 DataMonitor在构造函数中使用：
public 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);
}
对ZooKeeper.exists（）的调用检查znode的存在，设置一个watch，并传递一个引用本身（this）作为完成回调对象。 在这个意义上，它踢的东西，因为真正的处理发生在手表被触发。
注意
不要将完成回调与watch回调混淆。 当在服务器上完成watch操作的异步设置（通过ZooKeeper.exists（））时，会调用ZooKeeper.exists（）完成回调，
恰巧是在DataMonitor对象中实现的方法StatCallback.processResult（）。 另一方面，触发手表会向Executor对象发送一个事件，因为Executor注册为ZooKeeper对象的Watcher。 另外，您可能会注意到，DataMonitor也可以将自己注册为此特定监视事件的监视器。 这是ZooKeeper 3.0.0的新功能（多个监视器的支持）。 但是，在本示例中，DataMonitor不会注册为观察器。
当ZooKeeper.exists（）操作在服务器上完成时，ZooKeeper API在客户端上调用此完成回调：
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;
}
}
代码首先检查错误代码是否存在znode，致命错误和可恢复错误。 如果文件（或znode）存在，它从znode获取数据，然后调用Executor的exists（）回调，如果状态已更改。 注意，它不必对getData调用进行任何异常处理，因为它已监视挂起任何可能导致错误的事件：如果节点在调用ZooKeeper.getData（）之前被删除，ZooKeeper设置的watch事件 .exists（）触发回调; 如果出现通信错误，则在连接恢复时触发连接监视事件。 最后，请注意DataMonitor如何处理观看事件：
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:
//在这个特定的例子中，我们不需要在这里做任何事情 - 手表会自动重新注册服务器和任何手表触发，而客户端断开连接将交付（按顺序）
break;
case Expired:
// It's all over
dead = true;
listener.closing(KeeperException.Code.SessionExpired);
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);
}
}
如果客户端ZooKeeper库可以在会话到期（过期事件）之前重新建立与ZooKeeper的通信通道（SyncConnected事件），则所有会话的手表将自动与服务器重新建立（手表的自动重置是新的 ZooKeeper 3.0.0）。 有关更多信息，请参阅程序员指南中的ZooKeeper Watches。 在这个函数中位有点下降，当DataMonitor获取znode的事件时，它调用ZooKeeper.exists（）来找出发生了什么变化。
完成源代码列表
Executor.java
/**
* A simple example program to use DataMonitor to start and
* stop executables based on a znode. The program watches the
* specified znode and saves the data that corresponds to the
* znode in the filesystem. It also starts the specified program
* with the specified arguments when the znode exists and kills
* the program if the znode goes away.
*/
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream; import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper; public class Executor
implements Watcher, Runnable, DataMonitor.DataMonitorListener
{
String znode; DataMonitor dm; ZooKeeper zk; String filename; String exec[]; Process child; public 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);
} /**
* @param args
*/
public static void main(String[] args) {
if (args.length < 4) {
System.err
.println("USAGE: Executor hostPort znode filename program [args ...]");
System.exit(2);
}
String hostPort = args[0];
String znode = args[1];
String filename = args[2];
String exec[] = new String[args.length - 3];
System.arraycopy(args, 3, exec, 0, exec.length);
try {
new Executor(hostPort, znode, filename, exec).run();
} catch (Exception e) {
e.printStackTrace();
}
} /***************************************************************************
* We do process any events ourselves, we just need to forward them on.
*
* @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.proto.WatcherEvent)
*/
public void process(WatchedEvent event) {
dm.process(event);
} public void run() {
try {
synchronized (this) {
while (!dm.dead) {
wait();
}
}
} catch (InterruptedException e) {
}
} public void closing(int rc) {
synchronized (this) {
notifyAll();
}
} static class StreamWriter extends Thread {
OutputStream os; InputStream is; StreamWriter(InputStream is, OutputStream os) {
this.is = is;
this.os = os;
start();
} public void run() {
byte b[] = new byte[80];
int rc;
try {
while ((rc = is.read(b)) > 0) {
os.write(b, 0, rc);
}
} catch (IOException e) {
} }
} 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();
}
}
}
}
DataMonitor.java
/**
* A simple class that monitors the data and existence of a ZooKeeper
* node. It uses asynchronous ZooKeeper APIs.
*/
import java.util.Arrays; import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.data.Stat; public class DataMonitor implements Watcher, StatCallback { ZooKeeper zk; String znode; Watcher chainedWatcher; boolean dead; DataMonitorListener listener; byte prevData[]; public 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);
} /**
* Other classes use the DataMonitor by implementing this method
*/
public interface DataMonitorListener {
/**
* The existence status of the node has changed.
*/
void exists(byte data[]); /**
* The ZooKeeper session is no longer valid.
*
* @param rc
* the ZooKeeper reason code
*/
void closing(int rc);
} 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(KeeperException.Code.SessionExpired);
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);
}
} 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;
}
}
}
----------------------------------
http://zookeeper.apache.org/doc/r3.4.6/javaExample.html#sc_design