ZooKeeper Java 示例
一个简单的观察客户端
为了向您介绍 ZooKeeper Java API，我们在此开发了一个非常简单的观察客户端。此 ZooKeeper 客户端观察 znode 的更改，并通过启动或停止程序来响应。
要求
客户端有四个要求
- 它作为参数
- ZooKeeper 服务的地址
- znode 的名称 - 要观察的名称
- 要写入输出的文件的名称
- 带有参数的可执行文件。
- 它获取与 znode 关联的数据并启动可执行文件。
- 如果 znode 发生更改，客户端将重新获取内容并重新启动可执行文件。
- 如果 znode 消失，客户端将终止可执行文件。
程序设计
按照惯例，ZooKeeper 应用程序分为两个单元，一个用于维护连接，另一个用于监视数据。在此应用程序中，名为 Executor 的类维护 ZooKeeper 连接，名为 DataMonitor 的类监视 ZooKeeper 树中的数据。此外，Executor 包含主线程并包含执行逻辑。它负责少量用户交互，以及与您作为参数传入的可执行程序的交互，并且示例（根据要求）会根据 znode 的状态关闭并重新启动。
Executor 类
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) {
}
}
回想一下，执行程序的任务是启动和停止你在命令行中传入名称的可执行文件。它根据 ZooKeeper 对象触发的事件来执行此操作。正如你在上面的代码中所看到的，执行程序将对自身的一个引用作为 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 对象使用它与容器通信，该容器也是执行程序对象。DataMonitorListener 接口如下所示
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);
}
此接口在 DataMonitor 类中定义，并在 Executor 类中实现。当调用 Executor.exists() 时，执行程序根据要求决定启动还是关闭。回想一下，要求规定当 znode 不再存在时杀死可执行文件。
当调用 Executor.closing() 时，执行程序决定是否响应 ZooKeeper 连接永久消失而关闭自身。
正如你可能猜到的那样，DataMonitor 是响应 ZooKeeper 状态变化而调用这些方法的对象。
以下是执行程序对 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
对 ZooKeeper.exists() 的调用检查 znode 的存在，设置监视，并将对自身（this）的引用作为完成回调对象传递。从这个意义上说，它启动了所有事情，因为当监视被触发时，真正的处理才会发生。
注意
不要将完成回调与监视回调混淆。
ZooKeeper.exists()完成回调（恰好是 DataMonitor 对象中实现的方法StatCallback.processResult()）在异步设置监视操作（通过ZooKeeper.exists()）在服务器上完成时调用。另一方面，监视的触发向执行器对象发送事件，因为执行器已注册为 ZooKeeper 对象的监视器。
顺便说一下，您可能会注意到 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);</emphasis>
prevData = b;
}
}
该代码首先检查 znode 存在、致命错误和可恢复错误的错误代码。如果文件（或 znode）存在，它将从 znode 获取数据，然后在状态发生更改时调用执行器的 exists() 回调。请注意，它不必对 getData 调用执行任何异常处理，因为它已对可能导致错误的任何内容挂起监视：如果在调用 ZooKeeper.getData() 之前删除节点，则由 ZooKeeper.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:
// 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);
}
}
如果客户端 ZooKeeper 库可以在会话过期（已过期事件）之前重新建立与 ZooKeeper 的通信通道（同步连接事件），则会话的所有监视都将自动重新与服务器建立（监视的自动重置是 ZooKeeper 3.0.0 中的新功能）。有关此内容的更多信息，请参阅程序员指南中的 ZooKeeper 监视。在此函数的稍低位置，当 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;
}
}
}