网上有很多ZooKeeper的java客户端例子，我也看过很多，不过大部分写的都不好，有各种问题。兜兜转转还是觉得官方给的例子最为经典，在学习之余翻译下来，供朋友们参考
官网原文标题《ZooKeeper Java Example》
官网原文地址：http://zookeeper.apache.org/doc/current/javaExample.html#sc_completeSourceCode
针对本篇翻译文章，我还有一篇对应的笔记《ZooKeeper官方Java例子解读》，如果对官网文档理解有困难，可以结合我的笔记理解。
一个简单的监听客户端
通过开发一个非常简单的监听客户端，为你介绍ZooKeeper的Java API。此ZooKeeper的客户端，监听ZooKeeper中node的变化并做出响应。
需求
这个客户端有如下四个需求：
1、它接收如下参数：
- ZooKeeper服务的地址
- 被监控的znode的名称
- 可执行命令参数
2、它会取得znode上关联的数据，然后执行命令
3、如果znode变化，客户端重新拉取数据，再次执行命令
4、如果znode消失了，客户端杀掉进行的执行命令。
程序设计
一般我们会这么做，把ZooKeeper的程序分成两个单元，一个维护连接，另外一个监控数据。本程序中Executor类维护ZooKeeper的连接，DataMonitor监控ZooKeeper的数据。同时，Executor维护主线程以及执行逻辑。它负责对用户的交互做出响应，这里的交互既指根据你传入参数做出响应，也指根据znode的状态，关闭和重启。
Executor类
// 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(); } }DataMonitor类
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; } } }以上就是Java客户端开发案例ZooKeeper官方文档翻译的详细内容，更多关于java开发案例ooKeeper文档翻译的资料请关注0133技术站其它相关文章！
以上就是ZooKeeper官方文档之Java客户端开发案例翻译的详细内容，更多请关注0133技术站其它相关文章！
