时间：2022-10-12 10:04:20 | 栏目：JAVA代码 | 点击：次
官网原文标题《ZooKeeper Java Example》
官网原文地址：http://zookeeper.apache.org/doc/current/javaExample.html#sc_completeSourceCode
针对本篇翻译文章，我还有一篇对应的笔记《ZooKeeper官方Java例子解读》，如果对官网文档理解有困难，可以结合我的笔记理解。
通过开发一个非常简单的监听客户端，为你介绍ZooKeeper的Java API。此ZooKeeper的客户端，监听ZooKeeper中node的变化并做出响应。
这个客户端有如下四个需求：
1、它接收如下参数：
2、它会取得znode上关联的数据，然后执行命令
3、如果znode变化，客户端重新拉取数据，再次执行命令
4、如果znode消失了，客户端杀掉进行的执行命令。
一般我们会这么做，把ZooKeeper的程序分成两个单元，一个维护连接，另外一个监控数据。本程序中Executor类维护ZooKeeper的连接，DataMonitor监控ZooKeeper的数据。同时，Executor维护主线程以及执行逻辑。它负责对用户的交互做出响应，这里的交互既指根据你传入参数做出响应，也指根据znode的状态，关闭和重启。
// 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对象触发的事件来实现。就像上面的代码，在ZooKeeper的构造器中，Executor传递自己的引用作为watcher参数。同时，他传递自己的引用作为DataMonitorLisrener参数给DataMonitor构造器。在Executor定义中，实现了这些接口。
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...
ZooKeeper的Java API定义了Watcher接口。ZooKeeper用它来反馈给它的持有者。它仅支持一个方法process()，ZooKeeper用它来反馈主线程感兴趣的通用事件，例如ZooKeeper的连接状态，或者ZooKeeper session的状态。例子中的Executor只是简单的把事件传递给DataMonitor，由DataMonitor来决定怎么处理。为了方便，Executor或者其他的类似Executor的对象持有ZooKeeper连接，但是可以很自由的把事件委派给其他对象。它也用此作为触发watch事件的默认渠道。
public void process(WatchedEvent event) {
dm.process(event);
}
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不再存在时，杀掉进行中的执行命令。
当调用Executor.closing()，作为对ZooKeeper连接永久消失的响应，Executor决定是否关闭它自己。
就像你可能猜想的那样，，作为对ZooKeeper状态变化的响应，这些方法的调用者是DataMonitor。
下面是Exucutor中 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();
}
}
ZooKeeper的逻辑都在DataMonitor类中。他是异步和事件驱动的。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);
}
对zk.exists()的调用，会检查znode是否存在，设置watch，传递他自己的引用作为完成后的回调对象。这意味着，当watch被引发，真正的处理才开始。
Note
不要把完成回调和watch回调搞混。ZooKeeper.exists()完成时的回调，发生在DataMonitor对象实现的的StatCallback.processResult()方法中，调用发生在server上异步的watch设置操作（通过zk.exists()）完成时。
另一边，watch触发时，给Executor对象发送了一个事件，因为Executor注册成为ZooKeeper对象的一个watcher。
你可能注意到DataMonitor也可以注册它自己作为这个特定事件的watcher。这是ZooKeeper 3.0.0中加入的（多watcher的支持）。在这个例子中，DataMonitor并没有注册为watcher（译者：这里指zookeeper对象的watcher）。
当ZooKeeper.exists()在server上执行完成。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()前，node被删除了，通过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);
}
}
在session过期前，如果客户端zookeeper类库能重新发布和zookeeper的连接通道（SyncConnected event），session的所有watch将会重新发布。（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;
}
}
}