ZooKeeper Java 範例
簡單的監控用戶端
為了向您介紹 ZooKeeper Java API，我們在此開發了一個非常簡單的監控用戶端。這個 ZooKeeper 用戶端會監控 znode 的變更，並透過啟動或停止程式來回應。
需求
用戶端有四個需求
- 它採用下列參數
- ZooKeeper 服務的位址
- znode 的名稱 - 要監控的 znode
- 要寫入輸出的檔案名稱
- 帶有引數的可執行檔。
- 它會擷取與 znode 相關聯的資料並啟動可執行檔。
- 如果 znode 變更，用戶端會重新擷取內容並重新啟動可執行檔。
- 如果 znode 消失，用戶端會終止可執行檔。
程式設計
根據慣例，ZooKeeper 應用程式會分成兩個單元，一個維護連線，另一個監控資料。在此應用程式中，稱為 執行器 的類別維護 ZooKeeper 連線，而稱為 DataMonitor 的類別監控 ZooKeeper 樹狀結構中的資料。此外，執行器包含主執行緒和執行邏輯。它負責少量的使用者互動，以及與您傳入引數的可執行程式互動，而範例 (根據需求) 會根據 znode 的狀態關閉和重新啟動可執行程式。
執行器類別
執行器物件是範例應用程式的主要容器。它包含上述 程式設計 中所述的 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) {
}
}
回想一下，執行器的任務是啟動和停止您在命令列中傳入名稱的可執行檔。它會回應 ZooKeeper 物件觸發的事件來執行此操作。如您在上述程式碼中所見，執行器會將參考傳遞給它自己，作為 ZooKeeper 建構函式中的 Watcher 引數。它也會將參考傳遞給它自己，作為 DataMonitor 建構函式中的 DataMonitorListener 引數。根據執行器的定義，它實作了這兩個介面
public class Executor implements Watcher, Runnable, DataMonitor.DataMonitorListener {
...
Watcher 介面是由 ZooKeeper Java API 所定義的。ZooKeeper 使用它來與其容器進行通訊。它僅支援一個方法，process()，而 ZooKeeper 使用它來傳達主執行緒可能感興趣的通用事件，例如 ZooKeeper 連線狀態或 ZooKeeper 會話。此範例中的執行器僅將這些事件轉發至 DataMonitor，以決定如何處理它們。它這樣做的目的僅是為了說明，根據慣例，執行器或類似執行器的物件「擁有」ZooKeeper 連線，但它可以自由地將事件委派給其他物件。它也將其用作觸發監控事件的預設頻道。（稍後會詳細說明。）
public void process(WatchedEvent event) {
dm.process(event);
}
另一方面，DataMonitorListener 介面並非 ZooKeeper API 的一部分。它是一個完全自訂的介面，專為此範例應用程式所設計。DataMonitor 物件使用它來與其容器（也是執行器物件）進行通訊。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.exists() 時，執行器會根據需求決定啟動或關閉。請回想一下，需求表示當 znode 不再存在時，終止可執行檔。
當呼叫 Executor.closing() 時，執行器會決定是否要關閉本身，以回應 ZooKeeper 連線永久消失。
您可能已經猜到了，DataMonitor 是呼叫這些方法的物件，以回應 ZooKeeper 狀態的變更。
以下是執行器實作 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
呼叫 ZooKeeper.exists() 會檢查 znode 是否存在，設定監控，並傳遞對本身（this）的參考作為完成回呼物件。在這種意義上，它啟動了所有事情，因為真正的處理會在觸發監控時發生。
注意
不要將完成回呼與監控回呼混淆。
ZooKeeper.exists()完成回呼（碰巧是 DataMonitor 物件中實作的方法StatCallback.processResult()）會在非同步設定監控作業（由ZooKeeper.exists()執行）在伺服器上完成時呼叫。另一方面，觸發監控會將事件傳送至執行器物件，因為執行器已註冊為 ZooKeeper 物件的監控器。
順帶一提，您可能會注意到 DataMonitor 也可以將本身註冊為此特定監控事件的監控器。這是 ZooKeeper 3.0.0 的新功能（支援多個監控器）。不過，在此範例中，DataMonitor 並未註冊為監控器。
當 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);</emphasis>
prevData = b;
}
}
此程式碼會先檢查 znode 存在、致命錯誤和可復原錯誤的錯誤碼。如果檔案 (或 znode) 存在，它會從 znode 取得資料，然後在狀態變更時呼叫 Executor 的 exists() 回呼。請注意，它不必對 getData 呼叫執行任何例外處理，因為它已針對可能造成錯誤的任何事項掛起監視：如果在呼叫 ZooKeeper.getData() 之前節點已刪除，則 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);
}
}
如果用戶端 ZooKeeper 函式庫可以在會話過期 (Expired 事件) 之前重新建立與 ZooKeeper 的通訊頻道 (SyncConnected 事件)，則會話的所有監視都會自動與伺服器重新建立 (監視的自動重設是 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;
}
}
}