一个简单的 Watch Client
为了介绍 ZooKeeper Java API，我们在这里开发了一个非常简单的Watch Client。这个 ZooKeeper Client 监视 ZooKeeper 节点的变化，并通过启动或停止程序做出响应。
要求
Client 有四个要求：
- 采用以下参数：
- ZooKeeper 服务的地址
- znode 的名称 - 要监视的名称
- 将要输出写入的文件的名称
- 一个带参数的可执行文件。
- 获取与 znode 相关联的数据并启动可执行文件。
- 如果 znode 更改，Client 将重新获取内容并重新启动可执行文件。
- 如果 znode 消失，Client 会 kill 可执行文件。
程序设计
通常，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) {
}
}
回想一下，Executor 的任务是启动和停止在命令行上传递其名称的可执行文件。它响应由 ZooKeeper 对象发起的事件。正如你可以在上面的代码中看到的，Executor 传递一个引用自己作为 ZooKeeper 构造函数中的 Watcher 参数。它还将对自身的引用传递给 DataMonitor 构造函数的DataMonitorListener 参数。根据执行者的定义，它实现这两个接口：
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...
Watcher interface 由 ZooKeeper Java API 定义。 ZooKeeper 使用它来回传到它的容器。它只支持一个方法 process()，ZooKeeper 使用它来通信主线程将被插入的通用事件，例如 ZooKeeper 连接或 ZooKeeper 会话的状态。此示例中的执行器简单地将这些事件到 DataMonitor 来决定如何处理它们。它只是为了说明一点，按照惯例，执行器或一些类似执行器的对象 “拥有” ZooKeeper 连接，但它是可以将事件委托给其他事件到其他对象。它也使用它作为默认通道来触发监控事件。
public void process(WatchedEvent event) {
dm.process(event);
}
The DataMonitorListener interface, on the other hand, is not part of the the ZooKeeper API. It is a completely custom interface, designed for this sample application. The DataMonitor object uses it to communicate back to its container, which is also the the Executor object.The DataMonitorListener interface looks like this:
另一方面，DataMonitorListener 接口不是 ZooKeeper API 的一部分。它是一个为此示例程序设计的完全自定义的接口。 DataMonitor 对象使用它来回传到它的容器，也就是 Executor 对象。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() 被调用时，Executor 决定是否根据需求启动或关闭。回想一下，当 znode 挂了时，要求 kill 可执行文件。
当 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() 方法 。
另一方面，触发 watch 会向 Executor 对象发送事件，因为 Executor 注册为 ZooKeeper 对象的 Watcher。
另外，你可能会注意到，DataMonitor 也可以将自己注册为此特定 watch 事件的 Watcher 。这是 ZooKeeper 3.0.0 的新功能（支持多个 Watchers）。在本示例中，DataMonitor 不会注册为 Watcher 。
当服务器上完成 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 设置的 ZooKeeper.exists()触发回调；如果出现通信错误，则在连接恢复时触发连接监视事件。
最后，请注意 DataMonitor 如何处理 watch 事件：
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 的通信通道（SyncConnected 事件），则所有 watcher 将自动与服务器重新建立会话（手表的自动重置是新的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;
}
}
}
原文地址：https://zookeeper.apache.org/doc/trunk/javaExample.html
网友评论