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文档翻译的资料请关注脚本之家其它相关文章!