ZooKeeper官方文檔之Java客戶端開發(fā)案例翻譯
官網(wǎng)原文標(biāo)題《ZooKeeper Java Example》
官網(wǎng)原文地址：http://zookeeper.apache.org/doc/current/javaExample.html#sc_completeSourceCode
針對本篇翻譯文章，我還有一篇對應(yīng)的筆記《ZooKeeper官方Java例子解讀》，如果對官網(wǎng)文檔理解有困難，可以結(jié)合我的筆記理解。
一個簡單的監(jiān)聽客戶端
通過開發(fā)一個非常簡單的監(jiān)聽客戶端，為你介紹ZooKeeper的Java API。此ZooKeeper的客戶端，監(jiān)聽ZooKeeper中node的變化并做出響應(yīng)。
需求
這個客戶端有如下四個需求：
1、它接收如下參數(shù)：
- ZooKeeper服務(wù)的地址
- 被監(jiān)控的znode的名稱
- 可執(zhí)行命令參數(shù)
2、它會取得znode上關(guān)聯(lián)的數(shù)據(jù)，然后執(zhí)行命令
3、如果znode變化，客戶端重新拉取數(shù)據(jù)，再次執(zhí)行命令
4、如果znode消失了，客戶端殺掉進(jìn)行的執(zhí)行命令。
程序設(shè)計
一般我們會這么做，把ZooKeeper的程序分成兩個單元，一個維護(hù)連接，另外一個監(jiān)控數(shù)據(jù)。本程序中Executor類維護(hù)ZooKeeper的連接，DataMonitor監(jiān)控ZooKeeper的數(shù)據(jù)。同時，Executor維護(hù)主線程以及執(zhí)行邏輯。它負(fù)責(zé)對用戶的交互做出響應(yīng)，這里的交互既指根據(jù)你傳入?yún)?shù)做出響應(yīng)，也指根據(jù)znode的狀態(tài)，關(guān)閉和重啟。
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的工作是啟停通過命令行傳入的執(zhí)行命令。他通過響應(yīng)ZooKeeper對象觸發(fā)的事件來實現(xiàn)。就像上面的代碼，在ZooKeeper的構(gòu)造器中，Executor傳遞自己的引用作為watcher參數(shù)。同時，他傳遞自己的引用作為DataMonitorLisrener參數(shù)給DataMonitor構(gòu)造器。在Executor定義中，實現(xiàn)了這些接口。
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...ZooKeeper的Java API定義了Watcher接口。ZooKeeper用它來反饋給它的持有者。它僅支持一個方法process()，ZooKeeper用它來反饋主線程感興趣的通用事件，例如ZooKeeper的連接狀態(tài)，或者ZooKeeper session的狀態(tài)。例子中的Executor只是簡單的把事件傳遞給DataMonitor，由DataMonitor來決定怎么處理。為了方便，Executor或者其他的類似Executor的對象持有ZooKeeper連接，但是可以很自由的把事件委派給其他對象。它也用此作為觸發(fā)watch事件的默認(rèn)渠道。
public void process(WatchedEvent event) {
dm.process(event);
}DataMonitorListener接口，并不是ZooKeeper提供的API。它是為這個示例程序設(shè)計的自定義接口。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類實現(xiàn)。當(dāng)調(diào)用Executor.exists()，Executor根據(jù)需求決定是否啟動還是關(guān)閉?；貞浺幌拢枨筇岬疆?dāng)znode不再存在時，殺掉進(jìn)行中的執(zhí)行命令。
當(dāng)調(diào)用Executor.closing()，作為對ZooKeeper連接永久消失的響應(yīng)，Executor決定是否關(guān)閉它自己。
就像你可能猜想的那樣，，作為對ZooKeeper狀態(tài)變化的響應(yīng)，這些方法的調(diào)用者是DataMonitor。
下面是Exucutor中 DataMonitorListener.exists()和DataMonitorListener.closing()的實現(xiàn)
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類中。他是異步和事件驅(qū)動的。DataMonitor在構(gòu)造函數(shù)中完成啟動。
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()的調(diào)用，會檢查znode是否存在，設(shè)置watch，傳遞他自己的引用作為完成后的回調(diào)對象。這意味著，當(dāng)watch被引發(fā)，真正的處理才開始。
Note
不要把完成回調(diào)和watch回調(diào)搞混。ZooKeeper.exists()完成時的回調(diào)，發(fā)生在DataMonitor對象實現(xiàn)的的StatCallback.processResult()方法中，調(diào)用發(fā)生在server上異步的watch設(shè)置操作（通過zk.exists()）完成時。
另一邊，watch觸發(fā)時，給Executor對象發(fā)送了一個事件，因為Executor注冊成為ZooKeeper對象的一個watcher。
你可能注意到DataMonitor也可以注冊它自己作為這個特定事件的watcher。這是ZooKeeper 3.0.0中加入的（多watcher的支持）。在這個例子中，DataMonitor并沒有注冊為watcher（譯者：這里指zookeeper對象的watcher）。
當(dāng)ZooKeeper.exists()在server上執(zhí)行完成。ZooKeeper API將在客戶端發(fā)起這個完成回調(diào)
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存在返回的錯誤代碼，致命的錯誤及可恢復(fù)的錯誤。如果znode存在，將從znode取得數(shù)據(jù)，如果狀態(tài)發(fā)生改變，調(diào)用Executor的exists回調(diào)。不需要為getData做任何異常處理。因為它為任何可能引發(fā)錯誤的情況設(shè)置了監(jiān)控：如果在調(diào)用ZooKeeper.getData()前，node被刪除了，通過ZooKeeper.exists設(shè)置的監(jiān)聽事件被觸發(fā)回調(diào)；如果發(fā)生了通信錯誤，當(dāng)連接恢復(fù)時，連接的監(jiān)聽事件被觸發(fā)。
最后，看一下DataMonitor是如何處理監(jiān)聽事件的：
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類庫能重新發(fā)布和zookeeper的連接通道（SyncConnected event），session的所有watch將會重新發(fā)布。（zookeeper 3.0.0開始）。學(xué)習(xí)開發(fā)手冊中的ZooKeeper Watches。繼續(xù)往下講，當(dāng)DataMonitor從znode收到事件，他將會調(diào)用zookeeper.exists()，來找出發(fā)生了什么變化。
完整代碼清單
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客戶端開發(fā)案例ZooKeeper官方文檔翻譯的詳細(xì)內(nèi)容，更多關(guān)于java開發(fā)案例ooKeeper文檔翻譯的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要介紹了IDEA 快速返回上次查看代碼的位置的方法，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-08-08
本文主要介紹了maven配置本地倉庫的方法步驟，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-09-09
在Java的Spring框架中,將bean注入到容器中是核心概念之一,這是實現(xiàn)依賴注入的基礎(chǔ),Spring提供了多種方式來將bean注入到容器中,本文給大家總結(jié)了將bean注入到Spring中的幾種方式,需要的朋友可以參考下2023-12-12
這篇文章將用實例來和大家介紹一下entity轉(zhuǎn)換到vo的方法過程。文中的示例代碼講解詳細(xì)，對我們學(xué)習(xí)Java有一定的幫助，需要的可以參考一下2022-06-06
本篇文章主要就是分析Spring源碼剖析-Bean的實例化-屬性注入的相關(guān)知識，通過本文學(xué)習(xí)AbstractAutowireCapableBeanFactory#populateBean 方法的主要功能就是屬性填充，感興趣的朋友跟隨小編一起看看吧2021-06-06
這篇文章主要介紹了Java利用讀寫的方式實現(xiàn)音頻播放代碼實例,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2019-11-11
對arraylist中元素進(jìn)行排序?qū)嵗a
這篇文章主要介紹了對arraylist中元素進(jìn)行排序?qū)嵗a，還是比較不錯的，這里分享給大家，供需要的朋友參考。2017-11-11
本文主要介紹了java實現(xiàn)模擬USB接口的功能，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2022-07-07
在mybatis 中使用if else 進(jìn)行判斷的操作
這篇文章主要介紹了在mybatis 中使用if else 進(jìn)行判斷的操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-02-02
一個數(shù)如果恰好等于它的因子之和，這個數(shù)就稱為 "完數(shù) "。例如6=1＋2＋3.編程找出1000以內(nèi)的所有完數(shù)2017-02-02