本實例我們用的是java3.4.6版本，實例方便大家學(xué)習(xí)完后有不明白的可以在留言區(qū)討論。
開發(fā)應(yīng)用程序的zookeeper java綁定主要由兩個java包組成：
org.apache.zookeeper
org.apache.zookeeper.data
org.apache.zookeeper包由zookeeper監(jiān)視的接口定義和zookeeper的各種回調(diào)處理程序組成。 它定義了zookeeper客戶端類庫的主要類以及許多zookeeper事件類型和狀態(tài)的靜態(tài)定義。 org.apache.zookeeper.data包定義了與數(shù)據(jù)寄存器（也稱為znode）相關(guān)的特性，例如訪問控制列表（acl），ids，stats等。
zookeeper java api中的org.apache.zookeeper.server，org.apache.zookeeper.server.quorum和org.apache.zookeeper.server.upgrade包是服務(wù)器實現(xiàn)的一部分。 org.apache.zookeeper.client包用于查詢zookeeper服務(wù)器的狀態(tài)。
一 準(zhǔn)備開發(fā)環(huán)境
apache zookeeper是一個復(fù)雜的軟件，因此它需要運行許多其他類庫。 依賴庫作為jar文件在zookeeper發(fā)行版中附帶在lib目錄中。 核心zookeeper jar文件名字為zookeeper-3.4.6.jar，位于主目錄下。
要開發(fā)java的zookeeper應(yīng)用程序，我們必須設(shè)置指向zookeeper jar的類路徑，以及zookeeper所依賴的所有第三方庫。在 bin 目錄下有一個 zkenv.sh文件，可以用來設(shè)置classpath。
我們需要將腳本如下設(shè)置，在命令行中執(zhí)行以下語句：
|
1
2
|
$ zoobindir=${zk_home}/bin$ source ${zoobindir}/zkenv.sh |
shell變量zk_home被設(shè)置為安裝zookeeper的路徑，在我的設(shè)置中，它是/usr/share/zookeeper。 之后，classpath變量被正確設(shè)置，在我的系統(tǒng)中，如下所示：
|
1
2
|
$ echo $classpath /usr/share/zookeeper-3.4.6/bin/../build/classes :/usr/share/zookeeper-3.4.6/bin/../build/lib/*.jar :/usr/share/zookeeper-3.4.6/bin/../lib/slf4j-log4j12-1.6.1.jar :/usr/share/zookeeper-3.4.6/bin/../lib/slf4j-api-1.6.1.jar :/usr/share/zookeeper-3.4.6/bin/../lib/netty-3.7.0.final.jar :/usr/share/zookeeper-3.4.6/bin/../lib/log4j-1.2.16.jar :/usr/share/zookeeper-3.4.6/bin/../lib/jline-0.9.94.jar :/usr/share/zookeeper-3.4.6/bin/../zookeeper-3.4.6.jar :/usr/share/zookeeper-3.4.6/bin/../src/java/lib/*.jar :/usr/share/zookeeper-3.4.6/bin/../conf: |
在windows操作系統(tǒng)中，需要運行zkenv.cmd腳本。 現(xiàn)在可以使用classpath變量來編譯和運行使用zookeeper api編寫的java程序。 可以在uni/linux中的主目錄的.bashrc文件中找到zkenv.sh腳本，避免每次啟動shell會話時都采用它。
二 第一個zookeeper程序
為了引入zookeeper java api，讓我們從一個非常簡單的程序開始，它可以連接到localhost中的zookeeper實例，如果連接成功，它將在zookeeper名稱空間的根路徑下打印znode的列表。
這個程序的代碼如下所示：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
/*our first zookeeper program*/import java.io.ioexception;import java.util.arraylist;import java.util.list;import org.apache.zookeeper.keeperexception;import org.apache.zookeeper.zookeeper;public class hellozookeeper {public static void main(string[] args) throws ioexception {string hostport = "localhost:2181";string zpath = "/";list <string> zoochildren = new arraylist<string>();zookeeper zk = new zookeeper(hostport, 2000, null);if (zk != null) {try {zoochildren = zk.getchildren(zpath, false);system.out.println("znodes of '/': ");for (string child: zoochildren) {//print the childrensystem.out.println(child);}} catch (keeperexception e) {e.printstacktrace();} catch (interruptedexception e) {e.printstacktrace();}}}} |
在構(gòu)建和執(zhí)行前面的代碼片段之前，讓我們來看看它具體做了什么。代碼從導(dǎo)入語句開始。使用這些語句，我們導(dǎo)入了程序各個組件所需的包。如前所述，org.apache.zookeeper包包含客戶端與zookeeper服務(wù)器進(jìn)行交互所需的所有類和接口。在導(dǎo)入包之后，定義了一個名為hellozookeeper的類。由于我們要連接到在同一系統(tǒng)中運行的zookeeper實例，在main方法中將主機(jī)和端口字符串定義為localhost:2181。代碼行zk = new zookeeper(hostport, 2000, null)調(diào)用zookeeper構(gòu)造方法，該構(gòu)造方法嘗試連接到zookeeper服務(wù)器并返回一個引用。對于連接到zookeeper服務(wù)器實例并維護(hù)該連接的客戶端程序，需要維護(hù)一個實時會話。在此例中，構(gòu)造方法實例化的zk對象返回的引用表示這個會話。 zookeeper api是圍繞這個引用構(gòu)建的，每個方法調(diào)用都需要一個引用來執(zhí)行。
zookeeper類的構(gòu)造方法使用以下代碼創(chuàng)建zookeeper實例的引用：
|
1
|
zookeeper(string connectstring, int sessiontimeout, watcher watcher) |
使用的參數(shù)含義如下：
connectstring：以逗號分隔的主機(jī):端口號列表，每個對應(yīng)一個zookeeper服務(wù)器。 例如，10.0.0.1:2001，10.0.0.2:2002和10.0.0.3:2003表示三個節(jié)點的zookeeper ensemble的有效的主機(jī):端口匹配對。 sessiontimeout：這是以毫秒為單位的會話超時時間。這是zookeeper在宣布session結(jié)束之前，沒有從客戶端獲得心跳的時間。 watcher：一個watcher對象，如果創(chuàng)建，當(dāng)狀態(tài)改變和發(fā)生節(jié)點事件時會收到通知。這個watcher對象需要通過一個用戶定義的類單獨創(chuàng)建，通過實現(xiàn)watcher接口并將實例化的對象傳遞給zookeeper構(gòu)造方法。客戶端應(yīng)用程序可以收到各種類型的事件的通知，例如連接丟失、會話過期等。
zookeeper java api定義了另外帶有三個參數(shù)的構(gòu)造方法，以指定更高級的操作。代碼如下:
|
1
|
zookeeper(string connectstring, int sessiontimeout, watcher watcher, boolean canbereadonly) |
在zookeeper類的上面的構(gòu)造方法中，如果設(shè)置為true，boolean canbereadonly參數(shù)允許創(chuàng)建的客戶端在網(wǎng)絡(luò)分區(qū)的情況下進(jìn)入只讀模式。只讀模式是客戶端無法找到任何多數(shù)服務(wù)器的場景，但有一個可以到達(dá)的分區(qū)服務(wù)器，以只讀模式連接到它，這樣就允許對服務(wù)器的讀取請求，而寫入請求則不允許。客戶端繼續(xù)嘗試在后臺連接到大多數(shù)服務(wù)器，同時仍然保持只讀模式。分區(qū)服務(wù)器僅僅是zookeeper組的一個子集，它是由于集群中的網(wǎng)絡(luò)分配而形成的。大多數(shù)服務(wù)器構(gòu)成了ensemble中的大多數(shù)quorum。
以下構(gòu)造方法顯示了兩個附加參數(shù)的定義：
|
1
|
zookeeper(string connectstring, int sessiontimeout, watcher watcher, long sessionid, byte[] sessionpasswd) |
這個構(gòu)造方法允許zookeeper客戶端對象創(chuàng)建兩個額外的參數(shù)：
sessionid：在客戶端重新連接到zookeeper服務(wù)器的情況下，可以使用特定的會話id來引用先前連接的會話 sessionpasswd：如果指定的會話需要密碼，可以在這里指定
以下構(gòu)造方法是前兩個調(diào)用的組合：
|
1
|
zookeeper(string connectstring, int sessiontimeout,watcher watcher, long sessionid, byte[] sessionpasswd,boolean canbereadonly) |
此構(gòu)造方法是前兩個調(diào)用的組合，允許在啟用只讀模式的情況下重新連接到指定的會話。
note
zookeeper類的詳細(xì)java api文檔可以在http://zookeeper.apache.org/doc/r3.4.6/api/index.html上查詢。
現(xiàn)在，回到我們的zookeeper程序。 在調(diào)用構(gòu)造方法后，如果連接成功，我們將得到zookeeper服務(wù)器的引用。 我們通過下面的代碼將引用傳遞給getchildren方法：
|
1
|
zoochildren = zk.getchildren(zpath, false) |
zookeeper類的getchildren(string path，boolean watch)方法返回給定路徑上znode的子級列表。 我們只是迭代這個方法返回的列表，并將字符串打印到控制臺。
將程序命名為hellozookeeper.java，并編譯我們的程序如下：
|
1
|
$ javac -cp $classpath hellozookeeper.java |
在我們運行的程序之前，需要使用以下命令來啟動zookeeper服務(wù)器實例：
|
1
|
$ ${zk_home}/bin/zkserver.sh start |
運行程序如下：
|
1
|
$ java -cp $classpath hellozookeeper |
執(zhí)行程序會在控制臺上打印日志消息，顯示所使用的zookeeper版本，java版本，java類路徑，服務(wù)器體系結(jié)構(gòu)等等。 這里顯示了這些日志消息的一部分：
zookeeper java api生成的日志消息對調(diào)試非常有用。 它為我們提供了關(guān)于客戶端連接到zookeeper服務(wù)器，建立會話等后臺得信息。 上面顯示的最后三條日志消息告訴我們客戶端如何使用程序中指定的參數(shù)來啟動連接，以及在成功連接后，服務(wù)器如何為客戶端分配會話id。
最后，程序執(zhí)行最后在控制臺中輸出以下內(nèi)容：
我們可以使用zookeeper shell來驗證程序的正確性：
|
1
|
$ $zk_home/bin/zkcli.sh -server localhost |
恭喜！ 我們剛剛成功編寫了我們的第一個zookeeper客戶端程序。
二 實現(xiàn)watcher接口
zookeeper watcher監(jiān)視使客戶端能夠接收來自zookeeper服務(wù)器的通知，并在發(fā)生時處理這些事件。 zookeeper java api提供了一個名為watcher的公共接口，客戶端事件處理程序類必須實現(xiàn)該接口才能接收有關(guān)來自zookeeper服務(wù)器的事件通知。 以編程方式，使用這種客戶端的應(yīng)用程序通過向客戶端注冊回調(diào)(callback)對象來處理這些事件。
我們將實現(xiàn)watcher接口，處理與znode關(guān)聯(lián)的數(shù)據(jù)更改時由zookeeper生成的事件。
watcher接口在org.apache.zookeeper包中聲明如下：
|
1
2
3
|
public interface watcher {void process(watchedevent event);} |
為了演示znode數(shù)據(jù)監(jiān)視器（watcher），有兩個java類：datawatcher和dataupdater。 datawatcher將一直運行，并在/myconfig指定znode路徑中偵聽來自zookeeper服務(wù)器的nodedatachange事件。dataupdater類將定期更新此znode路徑中的數(shù)據(jù)字段，這將生成事件，并且在接收到這些事件后，datawatcher類將把更改后的數(shù)據(jù)打印到控制臺上。
以下是datawatcher.java類的代碼：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
import java.io.ioexception;import org.apache.zookeeper.createmode;import org.apache.zookeeper.keeperexception;import org.apache.zookeeper.watchedevent;import org.apache.zookeeper.watcher;import org.apache.zookeeper.zoodefs;import org.apache.zookeeper.zookeeper;public class datawatcher implements watcher, runnable {private static string hostport = "localhost:2181";private static string zoodatapath = "/myconfig";byte zoo_data[] = null;zookeeper zk;public datawatcher() {try {zk = new zookeeper(hostport, 2000, this);if (zk != null) {try {//create the znode if it doesn't exist, with the following code:if (zk.exists(zoodatapath, this) == null) {zk.create(zoodatapath, "".getbytes(), zoodefs.ids.open_acl_unsafe, createmode.persistent);}} catch (keeperexception | interruptedexception e) {e.printstacktrace();}}} catch (ioexception e) {e.printstacktrace();}}public void printdata() throws interruptedexception, keeperexception {zoo_data = zk.getdata(zoodatapath, this, null);string zstring = new string(zoo_data);// the following code prints the current content of the znode to the console:system.out.printf("\ncurrent data @ zk path %s: %s", zoodatapath, zstring);}@overridepublic void process(watchedevent event) {system.out.printf("\nevent received: %s", event.tostring());//we will process only events of type nodedatachangedif (event.gettype() == event.eventtype.nodedatachanged) {try {printdata();} catch (interruptedexception e) {e.printstacktrace();} catch (keeperexception e) {e.printstacktrace();}}}public static void main(string[] args)throws interruptedexception, keeperexception {datawatcher datawatcher = new datawatcher();datawatcher.printdata();datawatcher.run();}public void run() {try {synchronized (this) {while (true) {wait();}}} catch (interruptedexception e) {e.printstacktrace();thread.currentthread().interrupt();}}} |
我們來看一下datawatcher.java類的代碼來理解一個zookeeper監(jiān)視器的實現(xiàn)。 datawatcher公共類實現(xiàn)watcher接口以及runnable接口，打算將監(jiān)視器作為線程運行。 main方法創(chuàng)建datawatcher類的一個實例。 在前面的代碼中，datawatcher構(gòu)造方法嘗試連接到在本地主機(jī)上運行的zookeeper實例。 如果連接成功，我們用下面的代碼檢查znode路徑/myconfig是否存在：
|
1
|
if (zk.exists(zoodatapath, this) == null) { |
如果znode不存在zookeeper命名空間中，那么exists方法調(diào)用將返回null，并且嘗試使用代碼將其創(chuàng)建為持久化znode，如下所示：
|
1
|
zk.create(zoodatapath, "".getbytes(), zoodefs.ids.open_acl_unsafe, createmode.persistent); |
接下來是process方法，它在org.apache.zookeeper的watcher接口中聲明，并由datawatcher類使用以下代碼實現(xiàn)：
|
1
|
public void process(watchedevent event) { |
為了簡單起見，在process方法中，打印從zookeeper實例接收的事件，并僅對nodedatachanged類型的事件進(jìn)行進(jìn)一步處理，如下所示：
|
1
|
if (event.gettype() == event.eventtype.nodedatachanged) |
當(dāng)znode路徑/myconfig的數(shù)據(jù)字段發(fā)生任何更新或更改而收到nodedatachanged類型的事件時，調(diào)用printdata方法來打印znode的當(dāng)前內(nèi)容。 在znode上執(zhí)行一個getdata調(diào)用時，我們再次設(shè)置一個監(jiān)視，這是該方法的第二個參數(shù)，如下面的代碼所示：
|
1
|
zoo_data = zk.getdata(zoodatapath, this, null); |
監(jiān)視事件是發(fā)送給設(shè)置監(jiān)視的客戶端的一次性觸發(fā)器，為了不斷接收進(jìn)一步的事件通知，客戶端應(yīng)該重置監(jiān)視器。
dataupdater.java是一個簡單的類，它連接到運行本地主機(jī)的zookeeper實例，并用隨機(jī)字符串更新znode路徑/myconfig的數(shù)據(jù)字段。 在這里，我們選擇使用通用唯一標(biāo)識符（uuid）字符串更新znode，因為后續(xù)的uuid生成器調(diào)用將保證生成唯一的字符串。
dataupdater.java類代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
import java.io.ioexception;import java.util.uuid;import org.apache.zookeeper.keeperexception;import org.apache.zookeeper.watchedevent;import org.apache.zookeeper.watcher;import org.apache.zookeeper.zookeeper;public class dataupdater implements watcher {private static string hostport = "localhost:2181";private static string zoodatapath = "/myconfig";zookeeper zk;public dataupdater() throws ioexception {try {zk = new zookeeper(hostport, 2000, this);} catch (ioexception e) {e.printstacktrace();}}// updates the znode path /myconfig every 5 seconds with a new uuid string.public void run() throws interruptedexception, keeperexception {while (true) {string uuid = uuid.randomuuid().tostring();byte zoo_data[] = uuid.getbytes();zk.setdata(zoodatapath, zoo_data, -1);try {thread.sleep(5000); // sleep for 5 secs} catch(interruptedexception e) {thread.currentthread().interrupt();}}}public static void main(string[] args) throwsioexception, interruptedexception, keeperexception {dataupdater dataupdater = new dataupdater();dataupdater.run();}@overridepublic void process(watchedevent event) {system.out.printf("\nevent received: %s", event.tostring());}} |
上面的代碼使zookeeper服務(wù)器觸發(fā)一個nodedatachanged事件。 由于datawatcher為此znode路徑設(shè)置了監(jiān)視，因此它會接收數(shù)據(jù)更改事件的通知。 然后它檢索更新的數(shù)據(jù)，重置監(jiān)視，并在控制臺上打印數(shù)據(jù)。
使用以下命令編譯datawatcher和dataupdater類：
|
1
2
|
$ javac –cp $classpath datawatcher.java$ javac –cp $classpath dataupdater.java |
要執(zhí)行監(jiān)視器和更新程序，需要打開兩個終端窗口。 我要先運行監(jiān)視器，因為它創(chuàng)建了/myconfig的znode（如果還未在zookeeper的命名空間中創(chuàng)建的話）。 運行監(jiān)視器之前，請確保zookeeper服務(wù)器在本地主機(jī)上已經(jīng)運行。
在其中一個終端窗口中，通過運行以下命令來執(zhí)行watcher類：
|
1
|
$ java –cp $classpath datawatcher |
輸出類似于以下屏幕截圖所示的消息：
如前面的截圖所示，znode路徑/myconfig是由datawatcher類創(chuàng)建的。 它也打印znode的內(nèi)容，但沒有打印在控制臺中，因為我們在創(chuàng)建znode時沒有設(shè)置任何數(shù)據(jù)。 當(dāng)znode被創(chuàng)建時，類中的監(jiān)視者收到了nodecreated類型的事件通知，這個通知被打印在控制臺中。 datawatcher類繼續(xù)運行，并從zookeeper服務(wù)器偵聽/myconfig節(jié)點上的事件。
讓我們在另一個終端窗口中運行dataupdater類：
|
1
|
$ java -cp $classpath dataupdater |
將最初的zookeeper特定日志消息記錄到控制臺后，dataupdater類運行時沒有提示。 它將一個新的uuid字符串設(shè)置到zookeeper路徑/myconfig的數(shù)據(jù)字段中。 因此，看到每隔5秒鐘，在下面的屏幕截圖中顯示的輸出內(nèi)容打印在運行datawatche的終端窗口中：
datawatcher也可以使用zookeeper shell進(jìn)行測試。 繼續(xù)像以前一樣在終端中運行datawatcher類，并在另一個終端中調(diào)用zookeeper shell并運行以下屏幕截圖中所示的命令：
在datawatcher正在運行的終端中，將打印以下消息：
三 示例——群集監(jiān)視器
通過互聯(lián)網(wǎng)提供的流行服務(wù)，如電子郵件，文件服務(wù)平臺，在線游戲等，都是通過跨越多個數(shù)據(jù)中心的高度可用的成百上千臺服務(wù)器來服務(wù)的，而這些服務(wù)器通常在地理位置上分開。 在這種集群中，設(shè)置了一些專用的服務(wù)器節(jié)點來監(jiān)視生產(chǎn)網(wǎng)絡(luò)中承載服務(wù)或應(yīng)用程序的服務(wù)器的活躍性。 在云計算環(huán)境中，也用于管理云環(huán)境的這種監(jiān)控節(jié)點被稱為云控制器。 這些控制器節(jié)點的一個重要工作是實時檢測生產(chǎn)服務(wù)器的故障，并相應(yīng)地通知管理員，并采取必要的措施，例如將故障服務(wù)器上的應(yīng)用程序故障轉(zhuǎn)移到另一個服務(wù)器，從而確保容錯性和高可用性。
在本節(jié)中，我們將使用zookeeper java客戶端api開發(fā)一個簡約的分布式集群監(jiān)視器模型。 使用zookeeper的ephemeral znode概念來構(gòu)建這個監(jiān)視模型相當(dāng)簡單和優(yōu)雅，如以下步驟所述：
每個生產(chǎn)服務(wù)器運行一個zookeeper客戶端作為守護(hù)進(jìn)程。 這個過程連接到zookeeper服務(wù)器，并在/zookeeper命名空間的預(yù)定義路徑（比如/members）下創(chuàng)建一個帶有名稱（最好是其網(wǎng)絡(luò)名稱或主機(jī)名）的ephemeral znode。云控制器節(jié)點運行zookeeper監(jiān)視器進(jìn)程，該進(jìn)程監(jiān)視路徑/members并監(jiān)聽nodechildrenchanged類型的事件。 這個監(jiān)視器進(jìn)程作為服務(wù)或守護(hù)進(jìn)程運行，并設(shè)置或重置路徑上的監(jiān)視，并且實現(xiàn)其邏輯以調(diào)用適當(dāng)?shù)哪K來為監(jiān)視事件采取必要的行動。現(xiàn)在，如果生產(chǎn)服務(wù)器由于硬件故障或軟件崩潰而關(guān)閉，zookeeper客戶端進(jìn)程就會被終止，導(dǎo)致服務(wù)器和zookeeper服務(wù)之間的會話被終止。 由于ephemeral znode的屬性唯一，每當(dāng)客戶端連接關(guān)閉時，zookeeper服務(wù)會自動刪除路徑/members中的znode。路徑中znode的刪除引發(fā)了nodechildrenchanged事件，因此云控制器中的觀察器進(jìn)程會收到通知。 通過調(diào)用路徑/members中的getchildren方法，可以確定哪個服務(wù)器節(jié)點已經(jīng)關(guān)閉。然后，控制器節(jié)點可以采取適當(dāng)?shù)拇胧热鐖?zhí)行恢復(fù)邏輯以重啟另一臺服務(wù)器中的故障服務(wù)。這個邏輯可以構(gòu)建為實時工作，保證接近于零停機(jī)的時間和高度可用的服務(wù)。
為實現(xiàn)這個集群監(jiān)控模型，我們將開發(fā)兩個java類。 clustermonitor類將持續(xù)運行監(jiān)視器，以監(jiān)視zookeeper樹中的路徑/members。 處理完引發(fā)事件后，我們將在控制臺中打印znode列表并重置監(jiān)視。 另一個類clusterclient將啟動到zookeeper服務(wù)器的連接，在/members下創(chuàng)建一個ephemeral znode。
要模擬具有多個節(jié)點的集群，我們在同一臺計算機(jī)上啟動多個客戶端，并使用客戶端進(jìn)程的進(jìn)程id創(chuàng)建ephemeral znode。 通過查看進(jìn)程標(biāo)識，clustermonitor類可以確定哪個客戶進(jìn)程已經(jīng)關(guān)閉，哪些進(jìn)程還在。 在實際情況中，客戶端進(jìn)程通常會使用當(dāng)前正在運行的服務(wù)器的主機(jī)名創(chuàng)建ephemeral znode。 下面顯示了這兩個類的源代碼。
clustermonitor.java類定義如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
import java.io.ioexception;import java.util.list;import org.apache.zookeeper.createmode;import org.apache.zookeeper.keeperexception;import org.apache.zookeeper.watchedevent;import org.apache.zookeeper.watcher;import org.apache.zookeeper.zoodefs.ids;import org.apache.zookeeper.zookeeper;public class clustermonitor implements runnable {private static string membershiproot = "/members";private final watcher connectionwatcher;private final watcher childrenwatcher;private zookeeper zk;boolean alive=true;public clustermonitor(string hostport) throws ioexception, interruptedexception, keeperexception {connectionwatcher = new watcher() {@overridepublic void process(watchedevent event) {if(event.gettype()==watcher.event.eventtype.none && event.getstate() == watcher.event.keeperstate.syncconnected) {system.out.printf("\nevent received: %s", event.tostring());}}};childrenwatcher = new watcher() {@overridepublic void process(watchedevent event) {system.out.printf("\nevent received: %s", event.tostring());if (event.gettype() == event.eventtype.nodechildrenchanged) {try {//get current list of child znode, //reset the watchlist<string> children = zk.getchildren( membershiproot, this);wall("!!!cluster membership change!!!");wall("members: " + children);} catch (keeperexception e) {throw new runtimeexception(e);} catch (interruptedexception e) {thread.currentthread().interrupt();alive = false;throw new runtimeexception(e);}}}};zk = new zookeeper(hostport, 2000, connectionwatcher);// ensure the parent znode existsif(zk.exists(membershiproot, false) == null) {zk.create(membershiproot, "clustermonitorroot".getbytes(), ids.open_acl_unsafe, createmode.persistent);}// set a watch on the parent znodelist<string> children = zk.getchildren(membershiproot, childrenwatcher);system.err.println("members: " + children);}public synchronized void close() {try {zk.close();} catch (interruptedexception e) {e.printstacktrace();}}public void wall (string message) {system.out.printf("\nmessage: %s", message);}public void run() {try {synchronized (this) {while (alive) {wait();}}} catch (interruptedexception e) {e.printstacktrace();thread.currentthread().interrupt();} finally {this.close();}}public static void main(string[] args) throws ioexception, interruptedexception, keeperexception {if (args.length != 1) {system.err.println("usage: clustermonitor <host:port>");system.exit(0);}string hostport = args[0];new clustermonitor(hostport).run();}} |
clusterclient.java類定義如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
import java.io.ioexception;import java.lang.management.managementfactory;import org.apache.zookeeper.createmode;import org.apache.zookeeper.keeperexception;import org.apache.zookeeper.watchedevent;import org.apache.zookeeper.watcher;import org.apache.zookeeper.zoodefs.ids;import org.apache.zookeeper.zookeeper;public class clusterclient implements watcher, runnable {private static string membershiproot = "/members";zookeeper zk;public clusterclient(string hostport, long pid) {string processid = pid.tostring();try {zk = new zookeeper(hostport, 2000, this);} catch (ioexception e) {e.printstacktrace();}if (zk != null) {try {zk.create(membershiproot + '/' + processid, processid.getbytes(),ids.open_acl_unsafe, createmode.ephemeral);} catch (keeperexception | interruptedexception e) {e.printstacktrace();}}}public synchronized void close() {try {zk.close();}catch (interruptedexception e) {e.printstacktrace();}}@overridepublic void process(watchedevent event) {system.out.printf("\nevent received: %s", event.tostring());}public void run() {try {synchronized (this) {while (true) {wait();}}} catch (interruptedexception e) {e.printstacktrace();thread.currentthread().interrupt();} finally {this.close();}}public static void main(string[] args) {if (args.length != 1) {system.err.println("usage: clusterclient <host:port>");system.exit(0);}string hostport = args[0];//get the process idstring name = managementfactory.getruntimemxbean().getname();int index = name.indexof('@');long processid = long.parselong(name.substring(0, index));new clusterclient(hostport, processid).run();}} |
使用下面命令編譯這兩個類：
|
1
2
|
$ javac -cp $classpath clustermonitor.java$ javac -cp $classpath clusterclient.java |
要執(zhí)行群集監(jiān)控模型，打開兩個終端。 在其中一個終端中，運行clustermonitor類。 在另一個終端中，通過在后臺運行clusterclient類來執(zhí)行多個實例。
在第一個終端中，執(zhí)行clustermonitor類：
|
1
|
$ java -cp $classpath clustermonitorlocalhost:2181 |
如前面的示例所示，看到來自客戶端api的調(diào)試日志消息，最后，clustermonitor類開始監(jiān)視事件，輸入如下內(nèi)容：
現(xiàn)在，執(zhí)行clusterclient類的五個實例來模擬一個集群的五個節(jié)點。clusterclient在zookeeper樹的/members路徑中使用自己的進(jìn)程id創(chuàng)建ephemeral znode：
|
1
2
3
4
5
6
7
8
9
10
|
$ java -cp $classpath clusterclient localhost:2181 2>&1>/dev/null &[1] 4028$ java -cp $classpath clusterclient localhost:2181 2>&1>/dev/null &[2] 4045$ java -cp $classpath clusterclient localhost:2181 2>&1>/dev/null &[3] 4057$ java -cp $classpath clusterclient localhost:2181 2>&1>/dev/null &[4] 4072$ java -cp $classpath clusterclient localhost:2181 2>&1>/dev/null &[5] 4084 |
與此相對應(yīng)，將觀察到clustermonitor類檢測到這些新的clusterclient類實例，因為它正在監(jiān)視zookeeper樹的/members路徑上的事件。 這模擬了一個真正的集群中的節(jié)點加入事件。 可以在clustermonitor類的終端中看到輸出，這與下面的截圖中顯示的類似：
現(xiàn)在，如果殺死一個clusterclient.java進(jìn)程，那么它與zookeeper服務(wù)器一起維護(hù)的會話將被終止。因此，客戶端創(chuàng)建的ephemeral znode將被刪除。刪除將觸發(fā)nodechildrenchanged事件，該事件將被clustermonitor類捕獲。該模擬在集群中一個節(jié)點離開的場景。
讓我們用id 4084終止clusterclient進(jìn)程：
|
1
|
$ kill -9 4084 |
以下屏幕截圖顯示了clustermonitor類的終端中的輸出。 它列出了當(dāng)前可用的進(jìn)程及其進(jìn)程id，這些進(jìn)程id模擬了實時服務(wù)器：
上面的簡單而優(yōu)雅的集群監(jiān)控模型的示例實現(xiàn)展示了zookeeper的真正威力。 在沒有zookeeper的情況下，開發(fā)這樣一個能夠?qū)崟r監(jiān)控節(jié)點活躍度的模型將是一項真正的艱巨任務(wù)。
原文鏈接：https://www.cnblogs.com/IcanFixIt/p/7882107.html
