zookeeper理論知識(shí)VIP

1、簡(jiǎn)介ApacheZookeeper是由ApacheHadoop的Zookeeper子項(xiàng)目發(fā)展而來，現(xiàn)在已經(jīng)成為了Apache的頂級(jí)項(xiàng)目。Zookeeper為分布式系統(tǒng)提供了高效可靠且易于使用的協(xié)同服務(wù)，它可以為分布式應(yīng)用提供相當(dāng)多的服務(wù)，諸如統(tǒng)一命名服務(wù)，配置管理，狀態(tài)同步和組服務(wù)等。Zookeeper接口簡(jiǎn)單，開發(fā)人員不必過多地結(jié)結(jié)在分布式系統(tǒng)編程難于處理的同步和一致性問題上，你可以使用Zookeeper提供的現(xiàn)成（off-the-shelf）服務(wù)來實(shí)現(xiàn)分布式系統(tǒng)的配置管理，組管理，Leader選舉等功能。Zookeeper維護(hù)了大規(guī)模分布式系統(tǒng)中的常用對(duì)象，比如配置信息，層次化命名空間等，

2、本文將從開發(fā)者的角度詳細(xì)介紹Zookeeper的配置信息的意義以及Zookeeper的典型應(yīng)用場(chǎng)景（配置文件的管理、集群管理、分布式隊(duì)列、同步鎖、Leader選舉、隊(duì)列管理等）。Zookeeper安裝與配置本文采用Zookeeper-3.4.0以基礎(chǔ)介紹它的安裝步驟以及配置信息，最新的代碼可以到Zookeeper的官網(wǎng):工載。Zookeeper功能強(qiáng)大，但是安裝卻十分簡(jiǎn)單，下面重點(diǎn)以偽分布式模式來介紹Zookeeper的安裝。偽分布式模式安裝Zookeeper安裝模式包括：?jiǎn)螜C(jī)模式，偽分布式模式和完全的集群模式。單機(jī)模式最簡(jiǎn)單，本文將跳過單機(jī)模式安裝（單機(jī)模式安裝步驟參見Zeekeeper官方

3、文檔:/current/zookeeperStarted.html）,偽分布式模式與集群模式配置差別不大，由于手頭機(jī)器有限，所以本文采用了在單臺(tái)機(jī)器上偽分布式安裝。本文在Ubuntu12.04上操作，Java環(huán)境為OpenJDK1.7。安裝Zookeeper前首先下載你需要的版本，暫時(shí)解壓到指定目錄（本文解壓至/zookeeper/目錄下），并修改配置（可能需要多次修改配置文件），本次偽分布式模擬5個(gè)Zookeeper節(jié)點(diǎn)，事先在/tmpzookeeper目錄下建立5個(gè)文件夾，分別命名為:server001,server002,server003,server004,server005,然后在

4、每個(gè)serverOO#文件夾下面新建data和logs子文件夾。Zookeeper的配置文件主要在conf目錄，包括zoo.cfg（zoo_sample.cfg）和perties,修改zoo_sample.cfg,重命名為zoo.cgf,打開zoo.cfg,內(nèi)容如下：# ThenumberofmillisecondsofeachticktickTime=2000# Thenumberofticksthattheinitial#synchronizationphasecantakeinitLimit=10# Thenumberofticksthatcanpassbetween#

5、 sendingarequestandgettinganacknowledgementsyncLimit=5# thedirectorywherethesnapshotisstored.# donotuse/tmpforstorage,/tmphereisjust# examplesakes.dataDir=/tmp/zookeeper# theportatwhichtheclientswillconnectclientPort=2181# Besuretoreadthemaintenancesectionofthe# administratorguidebeforeturningonauto

6、purge.#/current/zookeeperAdmin.html#sc_maintenance# ThenumberofsnapshotstoretainindataDir#autopurge.snapRetainCount=3# Purgetaskintervalinhours# Setto"0"todisableautopurgefeature# autopurge.purgeInterval=1將內(nèi)容修改為（server001節(jié)點(diǎn)的配置文件）：# ThenumberofmillisecondsofeachticktickTime=2000# Thenumbero

7、fticksthattheinitial# synchronizationphasecantakeinitLimit=10# Thenumberofticksthatcanpassbetween# sendingarequestandgettinganacknowledgementsyncLimit=5# thedirectorywherethesnapshotisstored.# donotuse/tmpforstorage,/tmphereisjust# examplesakes.dataDir=/tmp/zookeeper/server001/datadataLogDir=/tmp/zo

8、okeeper/server001/logs# theportatwhichtheclientswillconnectclientPort=2181# Besuretoreadthemaintenancesectionofthe# administratorguidebeforeturningonautopurge.#/current/zookeeperAdmin.html#sc_maintenance#ThenumberofsnapshotstoretainindataDir#autopurge.snapRetainCount=3#Purgetaskintervalinhours#Setto

9、"0"todisableautopurgefeature#autopurge.purgeInterval=1server.1=server.2=server.3=server.4=server.5=?tickTime：這個(gè)時(shí)間是作為Zookeeper服務(wù)器之間或客戶端與服務(wù)器之間維持心跳的時(shí)間間隔，也就是每個(gè)tickTime時(shí)間就會(huì)發(fā)送一個(gè)心跳。?dataDir：顧名思義就是Zookeeper保存數(shù)據(jù)的目錄，默認(rèn)情況下，Zookeeper將寫數(shù)據(jù)的日志文件也保存在這個(gè)目錄里。?clientPort：這個(gè)端口就是客戶端連接Zookeeper服務(wù)器的端口，Zookeeper會(huì)監(jiān)

10、聽這個(gè)端口，接受客戶端的訪問請(qǐng)求。?initLimit：這個(gè)配置項(xiàng)是用來配置Zookeeper接受客戶端（這里所說的客戶端不是用戶連接Zookeeper服務(wù)器的客戶端，而是Zookeeper服務(wù)器集群中連接到Leader的Follower服務(wù)器）初始化連接時(shí)最長(zhǎng)能忍受多少個(gè)心跳時(shí)間間隔數(shù)。當(dāng)已經(jīng)超過5個(gè)心跳的時(shí)間（也就是tickTime）長(zhǎng)度后Zookeeper服務(wù)器還沒有收到客戶端的返回信息，那么表明這個(gè)客戶端連接失敗?？偟臅r(shí)間長(zhǎng)度就是5*2000=10秒?syncLimit：這個(gè)配置項(xiàng)標(biāo)識(shí)Leader與Follower之間發(fā)送消息，請(qǐng)求和應(yīng)答時(shí)間長(zhǎng)度，最長(zhǎng)不能超過多少個(gè)tickTime的時(shí)

11、間長(zhǎng)度，總的時(shí)間長(zhǎng)度就是2*2000=4秒?server.A=B：C：D：其中A是一個(gè)數(shù)字，表示這個(gè)是第幾號(hào)服務(wù)器；B是這個(gè)服務(wù)器的ip地址；C表示的是這個(gè)服務(wù)器與集群中的Leader服務(wù)器交換信息的端口；D表示的是萬一集群中的Leader服務(wù)器掛了，需要一個(gè)端口來重新進(jìn)行選舉，選出一個(gè)新的Leader,而這個(gè)端口就是用來執(zhí)行選舉時(shí)服務(wù)器相互通信的端口。如果是偽集群的配置方式，由于B都是一樣，所以不同的Zookeeper實(shí)例通信端口號(hào)不能一樣，所以要給它們分配不同的端口號(hào)。然后將此zookeeper包拷貝至/tmp/zookeeper/server001/目錄下，并在/tmp/zookeepe

12、r/server001/data/下建立一個(gè)myid文件，文件內(nèi)容為1，echo"1">>/tmp/zookeeper/server001/data/myid繼續(xù)修改/zookeeper/目錄中的zookeeper配置文件文件（server002的配置文件，注意clientPort=2182，與server001中的clientPort=2181不同，后續(xù)修改配置均需設(shè)置不同的clientPort），內(nèi)容如下：# ThenumberofmillisecondsofeachticktickTime=2000# Thenumberofticksthattheinit

13、ial# synchronizationphasecantakeinitLimit=10# Thenumberofticksthatcanpassbetween# sendingarequestandgettinganacknowledgementsyncLimit=5# thedirectorywherethesnapshotisstored.# donotuse/tmpforstorage,/tmphereisjust# examplesakes.dataDir=/tmp/zookeeper/server002/datadataLogDir=/tmp/zookeeper/server002

14、/logs# theportatwhichtheclientswillconnectclientPort=2182# Besuretoreadthemaintenancesectionofthe# administratorguidebeforeturningonautopurge.# /current/zookeeperAdmin.html#sc_maintenance# ThenumberofsnapshotstoretainindataDir# autopurge.snapRetainCount=3# Purgetaskintervalinhours# Setto"0"

15、;todisableautopurgefeature# autopurge.purgeInterval=1server.1=server.2=server.3=server.4=server.5=然后將此zookeeper包拷貝至/tmp/zookeeper/server002/目錄下，并在/tmp/zookeeper/server002/data/下建立一個(gè)myid文件，文件內(nèi)容為2，echo"2">>/tmp/zookeeper/server001/data/myid依次修改配置文件，建立server003，server004，server005節(jié)點(diǎn)文件夾，

16、完成上述步驟后/tmp/zookeeper目錄結(jié)構(gòu)如下：forhappyforhappy-lenovo:/tmp/zookeeper$tree-d-L2server001IdataIlogs1zookeeper-3.4.0server002I|dataIIlogsI1zookeeper-3.4.0Iserver003IIdataIIlogsI1zookeeper-3.4.0Iserver004IIdataIIlogsI1zookeeper-3.4.01 server005IdataIlogs1 zookeeper-3.4.0然后依次進(jìn)入每個(gè)文件夾節(jié)點(diǎn)的zookeeper目錄中，啟動(dòng)zookee

17、per服務(wù)，$bin/zkServer.shstart如果一切順利，Zookeeper偽分布式模式安裝成功，下面驗(yàn)證Zookeeper安裝的正確性。進(jìn)入任意一個(gè)文件夾節(jié)點(diǎn)的zookeeper包所在的目錄，執(zhí)行一下命令：$bin/zkCli.sh-server執(zhí)行成功后：forhappyforhappy-lenovo:/tmp/zookeeper/server001/zookeeper-bin/zkCli.sh-serverConnectingtoWelcometoZooKeeper!WATCHER:WatchedEventstate:SyncConnectedtype:Nonepath:nul

18、lzk:0help幫助：zk:0helpZooKeeper-serverhost:portcmdargsconnecthost:portgetpathwatchlspathwatchsetpathdataversionrmrpathdelquota-n|-bpathquitprintwatcheson|offcreate卜s-epathdataaclstatpathwatchclosels2pathwatchhistorylistquotapathsetAclpathaclgetAclpathsyncpathredocmdnoaddauthschemeauthdeletepathversion

19、setquota-n|-bvalpath至此，Zookeeper安裝完成，下一篇博客將介紹ZookeeperJavaAPI,并給出Zookeeper典型的應(yīng)用場(chǎng)景。簡(jiǎn)介ApacheZookeeper是由ApacheHadoop的Zookeeper子項(xiàng)目發(fā)展而來，現(xiàn)在已經(jīng)成為了Apache的頂級(jí)項(xiàng)目。Zookeeper為分布式系統(tǒng)提供了高效可靠且易于使用的協(xié)同服務(wù)，它可以為分布式應(yīng)用提供相當(dāng)多的服務(wù)，諸如統(tǒng)一命名服務(wù)，配置管理，狀態(tài)同步和組服務(wù)等。Zookeeper接口簡(jiǎn)單，開發(fā)人員不必過多地結(jié)結(jié)在分布式系統(tǒng)編程難于處理的同步和一致性問題上，你可以使用Zookeeper提供的現(xiàn)成(off-the

20、-shelf)服務(wù)來實(shí)現(xiàn)分布式系統(tǒng)的配置管理，組管理，Leader選舉等功能。Zookeeper維護(hù)了大規(guī)模分布式系統(tǒng)中的常用對(duì)象，比如配置信息，層次化命名空間等，本文將從開發(fā)者的角度詳細(xì)介紹Zookeeper的配置信息的意義以及Zookeeper的典型應(yīng)用場(chǎng)景(配置文件的管理、集群管理、分布式隊(duì)列、同步鎖、Leader選舉、隊(duì)列管理等)。上一篇博客主要講了ApacheZookeeper的安裝與配置，本文主要介紹ZookeeperJavaAPI。Zookeeper提供了原生的JavaAPI,另外還有C調(diào)用接口，本文暫不介紹Zookeeper的CAPI。ZookeeperJavaAPIZooke

21、eper作為一個(gè)分布式服務(wù)框架，主要用來解決分布式集群中應(yīng)用系統(tǒng)的一致性問題，它能提供基于類似于文件系統(tǒng)的目錄節(jié)點(diǎn)樹方式的數(shù)據(jù)存儲(chǔ)，但是Zookeeper并不是用來專門存儲(chǔ)數(shù)據(jù)的，它的作用主要是用來維護(hù)和監(jiān)控你存儲(chǔ)的數(shù)據(jù)的狀態(tài)變化。通過監(jiān)控這些數(shù)據(jù)狀態(tài)的變化，從而可以達(dá)到基于數(shù)據(jù)的集群管理。Zookeeper客戶端在連接Zookeeper服務(wù)器需要實(shí)例化一個(gè)，然后調(diào)用該類提供的接口與Zookeeper服務(wù)器進(jìn)行交互。如果不指明，該類的所有方法均是線程安全的。一旦Zookeeper客戶端與服務(wù)器建立連接，客戶端就會(huì)被分配一個(gè)會(huì)話ID(sessionID),客戶端會(huì)定期向服務(wù)器端發(fā)送心跳以保持該會(huì)

22、話有效。只要客戶端會(huì)話有效，應(yīng)用程序可以調(diào)用Zookeeper客戶端的接口與服務(wù)器端進(jìn)行交互。下表主要介紹了ZooKeeper方法(該表摘自IBMDeveloperworks:分布式服務(wù)框架Zookeeper-管理分布式環(huán)境中的數(shù)據(jù))方法名String create (Stringpath, byte口data,List< ACL> acl, CreateMode createMode)Stat exists (String path,boolean watch)Stat exists (String path, Watcher watcher)void delete (Strin

23、g path, int version)List <String > getChildren (String path, boolean watch)Stat setData (String path, byte data, int version)方法功能描述創(chuàng)建一個(gè)給定的目錄節(jié)點(diǎn)path,并給它設(shè)置數(shù)據(jù)，CreateMode標(biāo)識(shí)有四種形式的目錄節(jié)點(diǎn)，分別是PERSISTENT持久化目錄節(jié)點(diǎn)，這個(gè)目錄節(jié)點(diǎn)存儲(chǔ)不會(huì)丟失;PERSISTENT_SEQUENT:IAL序自動(dòng)編三節(jié)點(diǎn)，這種目錄節(jié)點(diǎn)會(huì)根據(jù)當(dāng)前已近存在的節(jié)點(diǎn)數(shù)1,然后返回給客戶端已經(jīng)成功創(chuàng)建的目錄節(jié)點(diǎn)名；EPHEMERAL臨

24、時(shí)目錄節(jié)點(diǎn),一旦創(chuàng)建這個(gè)節(jié)點(diǎn)的客服務(wù)器端口也就是session超時(shí)，這種節(jié)點(diǎn)會(huì)被自除；EPHEMERAL_SEQUENTIAW自動(dòng)編號(hào)節(jié)點(diǎn)。判斷某個(gè)path是否存在，并設(shè)置是否監(jiān)控這個(gè)目這里的watcher是在創(chuàng)建ZooKeeper實(shí)例時(shí)指定watcher,exists方法還有一個(gè)重載方法,可以指Iwatcher。重載方法，這里給某個(gè)目錄節(jié)點(diǎn)設(shè)置特定的watcWatcher在ZooKeeper是一個(gè)核心功能，Watche控目錄節(jié)點(diǎn)的數(shù)據(jù)變化以及子目錄的變化，一旦這發(fā)生變化，服務(wù)器就會(huì)通知所有設(shè)置在這個(gè)目錄節(jié)Watcher,從而每個(gè)客戶端都很快知道它所關(guān)注的目的狀態(tài)發(fā)生變化，而做出相應(yīng)的反應(yīng)。刪

25、除path對(duì)應(yīng)的目錄節(jié)點(diǎn)，version為-1可力何版本，也就刪除了這個(gè)目錄節(jié)點(diǎn)所有數(shù)據(jù)。獲取指定path下的所有子目錄節(jié)點(diǎn)，同樣getCI方法也有一個(gè)重載方法可以設(shè)置特定的watcher.節(jié)點(diǎn)的狀態(tài)。給path設(shè)置數(shù)據(jù)，可以指定這個(gè)數(shù)據(jù)的版本號(hào)，version為-1怎可以匹配任何版本。byte口getData(Stringpath,booleanwatch,獲取這個(gè)path對(duì)應(yīng)的目錄節(jié)點(diǎn)存儲(chǔ)的數(shù)據(jù)，數(shù)據(jù)StatstOt)等信息可以通過stat來指定，同時(shí)還可以設(shè)置是一這個(gè)目錄節(jié)點(diǎn)數(shù)據(jù)的狀態(tài)。voidaddAuthInfo(Stringscheme,byte口auth)客戶端將自己的授權(quán)信息提

26、交給服務(wù)器，服務(wù)器將個(gè)授權(quán)信息驗(yàn)證客戶端的訪問權(quán)限。StatsetACL(Stringpath,List<ACLacl,給某個(gè)目錄節(jié)點(diǎn)重新設(shè)置訪問權(quán)限，需要注意的是intversion)Zookeeper中的目錄節(jié)點(diǎn)權(quán)限不具有傳遞性，父目的權(quán)限不能傳遞給子目錄節(jié)點(diǎn)。目錄節(jié)點(diǎn)ACL由i成：perms和id。Perms有ALL、READWRITECREATEDELETE/種而id標(biāo)識(shí)了訪問目錄節(jié)點(diǎn)的身份列表，默認(rèn)情況下兩種：ANYONE_ID_UNSAFE=newId("world","anyone")AUTH_IDS=newId("auth

27、","")分別表示任何人訪問和創(chuàng)建者擁有訪問權(quán)限。List<ACLgetACL(Stringpath,Statstat)獲取某個(gè)目錄節(jié)點(diǎn)的訪問權(quán)限列表除了以上這些上表中列出的方法之外還有一些重載方法，如都提供了一個(gè)回調(diào)類的重載方法以及可以設(shè)置特定Watcher的重載方法，具體的方法可以參考ZooKeeper類的API說明。下面給出JavaAPI的基本操作：/創(chuàng)建一個(gè)與服務(wù)器的連接ZooKeeperzk=newZooKeeper("localhost:"+CLIENT_PORT,ClientBase.CONNECTION_TIMEOUT,n

28、ewWatcher()/監(jiān)控所有被觸勺事件publicvoidprocess(WatchedEventevent)"已經(jīng)觸發(fā)了"+event.getType()+"事件！");)；/創(chuàng)建一個(gè)目錄節(jié)點(diǎn)zk.create("/testRootPath","testRootData".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);/創(chuàng)建一個(gè)子目錄節(jié)點(diǎn)zk.create("/testRootPath/testChildPathOne",&qu

29、ot;testChildDataOne".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);String(zk.getData("/testRootPath",false,null);/取出子目錄節(jié)點(diǎn)列表"/testRootPath",true);/修改子目錄節(jié)點(diǎn)數(shù)據(jù)zk.setData("/testRootPath/testChildPathOne","modifyChildDataOne".getBytes(),-i);"目錄節(jié)點(diǎn)狀態(tài)

30、："+zk.exists("/testRootPath",true)+"");/創(chuàng)建另外一個(gè)子目錄節(jié)點(diǎn)zk.create("/testRootPath/testChildPathTwo","testChildDataTwo".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);String(zk.getData("/testRootPath/testChildPathTwo",true,null);/刪除子目錄節(jié)點(diǎn)zk.delet

31、e("/testRootPath/testChildPathTwo",-1);zk.delete("/testRootPath/testChildPathOne",-1);/刪除父目錄節(jié)點(diǎn)zk.delete("/testRootPath",-1);/關(guān)閉連接zk.close();簡(jiǎn)介ApacheZookeeper是由ApacheHadoop的Zookeeper子項(xiàng)目發(fā)展而來，現(xiàn)在已經(jīng)成為了Apache的頂級(jí)項(xiàng)目。Zookeeper為分布式系統(tǒng)提供了高效可靠且易于使用的協(xié)同服務(wù)，它可以為分布式應(yīng)用提供相當(dāng)多的服務(wù)，諸如統(tǒng)一命名服務(wù)，配置

32、管理，狀態(tài)同步和組服務(wù)等。Zookeeper接口簡(jiǎn)單，開發(fā)人員不必過多地結(jié)結(jié)在分布式系統(tǒng)編程難于處理的同步和一致性問題上，你可以使用Zookeeper提供的現(xiàn)成(off-the-shelf)服務(wù)來實(shí)現(xiàn)分布式系統(tǒng)的配置管理，組管理，Leader選舉等功能。英文原文地址：/current/javaExample.html一個(gè)簡(jiǎn)單的ZookeeperWatch客戶端為了介紹ZookeeperJavaAPI的基本用法，本文將帶你如何一步一步實(shí)現(xiàn)一個(gè)功能簡(jiǎn)單的Zookeeper客戶端。該Zookeeper客戶端會(huì)監(jiān)視一個(gè)你指定Zookeeper節(jié)點(diǎn)Znode,當(dāng)被監(jiān)視的節(jié)點(diǎn)發(fā)生變化時(shí)，客戶端會(huì)啟動(dòng)或者停

33、止某一程序?；疽笤摽蛻舳司邆渌膫€(gè)基本要求：?客戶端所帶參數(shù)：oZookeeper服務(wù)地址。o被監(jiān)視的Znode節(jié)點(diǎn)名稱。o可執(zhí)行程序及其所帶的參數(shù)?客戶端會(huì)獲取被監(jiān)視Znode節(jié)點(diǎn)的數(shù)據(jù)并啟動(dòng)你所指定的可執(zhí)行程序。?如果被監(jiān)視的Znode節(jié)點(diǎn)發(fā)生改變，客戶端重新獲取其內(nèi)容并再次啟動(dòng)你所指定的可執(zhí)行程序。?如果被監(jiān)視的Znode節(jié)點(diǎn)消失，客戶端會(huì)殺死可執(zhí)行程序。程序設(shè)計(jì)一般而言，Zookeeper應(yīng)用程序分為兩部分，其中一部分維護(hù)與服務(wù)器端的連接，另外一部分監(jiān)視Znode節(jié)點(diǎn)的數(shù)據(jù)。在本程序中，Executor類負(fù)責(zé)維護(hù)Zookeeper連接，DataMonitor類監(jiān)視Zookeeper目

34、錄樹中的數(shù)據(jù)，同時(shí)，Executor包含了主線程和程序主要的執(zhí)行邏輯，它負(fù)責(zé)少量的用戶交互，以及與可執(zhí)行程序的交互，該可執(zhí)行程序接受你向它傳入的參數(shù)，并且會(huì)根據(jù)被監(jiān)視的Znode節(jié)點(diǎn)的狀態(tài)變化停止或重啟。Executor類Executor對(duì)象是本例程最基本的“容器”，它包括Zookeeper對(duì)象和DataMonitor對(duì)象。publicstaticvoidmain(Stringargs)if(args.length<4)System.err.println("USAGE:ExecutorhostPortznodefilenameprogramargs.");Syste

35、m.exit(2);StringhostPort=args0;Stringznode=args1;Stringfilename=args2;Stringexec=newStringargs.length-3;System.arraycopy(args,3,exec,0,exec.length);trynewExecutor(hostPort,znode,filename,exec).run();catch(Exceptione)e.printStackTrace();publicExecutor(StringhostPort,Stringznode,Stringfilename,Stringe

36、xec)throwsKeeperException,IOExceptionthis.filename=filename;this.exec=exec;zk=newZooKeeper(hostPort,3000,this);dm=newDataMonitor(zk,znode,null,this);publicvoidrun()trysynchronized(this)while(!dm.dead)wait();catch(InterruptedExceptione)回憶一下Executor的任務(wù)是根據(jù)Zookeeper中Znode節(jié)點(diǎn)狀態(tài)改變所觸發(fā)的事件來啟動(dòng)和停止你在命令行指定的可執(zhí)行程序，

37、在上面的代碼你可以看到，Executor類在其構(gòu)造函數(shù)中實(shí)例化Zookeeper對(duì)象時(shí)，將其自身的引用作為Watch參數(shù)傳遞給Zookeeper的構(gòu)造函數(shù)，同時(shí)它也將其自身的引用作為DataMonitorListener參數(shù)傳遞給DataMonitor的構(gòu)造函數(shù)。Executor本身實(shí)現(xiàn)了以下接口：publicclassExecutorimplementsWatcher,Runnable,DataMonitor.DataMonitorListenerWatcher接口是在ZooKeeperJavaAPI中定義的。ZooKeeper用它來與“容器”(此處“容器”與上面的Executor類相似)進(jìn)

38、行通信，Watcher只支持一個(gè)方法，即process(),ZooKeeper用該函數(shù)來處理主線程可能感興趣的事件，例如Zookeeper連接或會(huì)話的狀態(tài)，本例中的“容器”Executor只是簡(jiǎn)單地把事件向下傳遞給DataMonitor，具體如何處理事件是由DataMonitor決定的。本文只是簡(jiǎn)單地描述了如何使用Watcher，通常情況下，Executor或與Executor類似的對(duì)象擁有與Zookeeper服務(wù)端的連接，但它可以將事件傳遞給其他對(duì)象，并有其它的對(duì)象處理該事件。publicvoidprocess(WatchedEventevent)cess(event);Data

39、MonitorListener接口本身不是ZookeeperAPI的一部分，它完全是一個(gè)自定義的接口，可以說是專門為本程序設(shè)計(jì)的。DataMonitor對(duì)象使用該接口和“容器”(即Executor類)進(jìn)行通信，DataMonitorListener接口如下：publicinterfaceDataMonitorListener/* Theexistencestatusofthenodehaschanged.*/voidexists(bytedata);/* TheZooKeepersessionisnolongervalid.* paramrc* theZooKeeperreasoncode*/

40、voidclosing(intrc);該接口在DataMonitor中定義，Executor類實(shí)現(xiàn)該接口，當(dāng)Executor.exists()被調(diào)用的時(shí)候，Executor決定是否啟動(dòng)或停止事先指定的應(yīng)用程序(回憶一下前文所說的，當(dāng)Znode消失時(shí)Zookeeper客戶端會(huì)殺死該可執(zhí)行程序)。當(dāng)Executor.closing()被調(diào)用的時(shí)候，Executor會(huì)根據(jù)Zookeeper連接永久性地消失來決定是否關(guān)閉自己。你或許已經(jīng)猜到，DataMonitor對(duì)象根據(jù)Zookeeper狀態(tài)變化來調(diào)用這些方法吧？以下是Executor類中實(shí)現(xiàn)DataMonitorListener.exists()和

41、DataMonitorListener.closing()的代碼：publicvoidexists(bytedata)if(data=null)if(child!=null)"Killingprocess");child.destroy();trychild.waitFor();catch(InterruptedExceptione)child=null;elseif(child!=null)"Stoppingchild");child.destroy();trychild.waitFor();catch(InterruptedExceptione)e.

42、printStackTrace();tryFileOutputStreamfos=newFileOutputStream(filename);fos.write(data);fos.close();catch(IOExceptione)e.printStackTrace();try"Startingchild");child=Runtime.getRuntime().exec(exec);newStreamWriter(child.getInputStream(),System.out);newStreamWriter(child.getErrorStream(),Syst

43、em.err);catch(IOExceptione)e.printStackTrace();publicvoidclosing(intrc)synchronized(this)notifyAll();DataMonitor類DataMonitor類是本程序Zookeeper邏輯的核心，它差不多是異步的，并由事件驅(qū)動(dòng)的。DataMonitor構(gòu)造函數(shù)如下：publicDataMonitor(ZooKeeperzk,Stringznode,WatcherchainedWatcher,DataMonitorListenerlistener)this.zk=zk;this.znode=znode;t

44、his.chainedWatcher=chainedWatcher;this.listener=listener;/Getthingsstartedbycheckingifthenodeexists.Wearegoing/tobecompletelyeventdrivenzk.exists(znode,true,this,null);調(diào)用ZooKeeper.exists()檢查指定的Znode是否存在，并設(shè)置監(jiān)視，傳遞自身引用作為回調(diào)對(duì)象，在某種意義上，在watch觸發(fā)時(shí)就會(huì)引起真實(shí)的處理流程。當(dāng)ZooKeeper.exists()操作在服務(wù)器端完成時(shí)，ZooKeeperAPI會(huì)在客戶端調(diào)用c

45、ompletioncallback：publicvoidprocessResult(intrc,Stringpath,Objectctx,Statstat)booleanexists;switch(rc)caseCode.Ok:exists=true;break;caseCode.NoNode:exists=false;break;caseCode.SessionExpired:caseCode.NoAuth:dead=true;listener.closing(rc);return;default:/Retryerrorszk.exists(znode,true,this,null);ret

46、urn;byteb=null;if(exists)tryb=zk.getData(znode,false,null);catch(KeeperExceptione)/Wedon'tneedtoworryaboutrecoveringnow.Thewatch/callbackswillkickoffanyexceptionhandlinge.printStackTrace();catch(InterruptedExceptione)return;if(b=null&&b!=prevData)|(b!=null&&!Arrays.equals(prevDat

47、a,b)listener.exists(b);prevData=b;上述代碼首先檢查Znode是否存在，以及其他重大的不可恢復(fù)的錯(cuò)誤。如果文件(或者Znode)存在，它將從Znode獲取數(shù)據(jù)，如果狀態(tài)發(fā)生變化再調(diào)用Executor的exists()回調(diào)函數(shù)。注意，getData函數(shù)本省必須要做任何的異常處理，因?yàn)楸旧砭陀斜O(jiān)視可以處理任何錯(cuò)誤：如果節(jié)點(diǎn)在調(diào)用ZooKeeper.getData()之前被刪除，ZooKeeper.exists()就會(huì)觸發(fā)回調(diào)函數(shù)，如果存在通信錯(cuò)誤，在連接上的監(jiān)視會(huì)在該連接重建之前觸發(fā)相應(yīng)的事件，同時(shí)引發(fā)相應(yīng)的處理。最后，DataMonitor處理監(jiān)視事件的代碼如下：

48、publicvoidprocess(WatchedEventevent)Stringpath=event.getPath();if(event.getType()=/Wearearebeingtoldthatthestateofthe/connectionhaschangedswitch(event.getState()caseSyncConnected:/Inthisparticularexamplewedon'tneedtodoanything/here-watchesareautomaticallyre-registeredwith/serverandanywatchestrig

49、geredwhiletheclientwas/disconnectedwillbedelivered(inorderofcourse)break;caseExpired:/It'salloverdead=true;listener.closing(;break;elseif(path!=null&&path.equals(znode)/Somethinghaschangedonthenode,let'sfindoutzk.exists(znode,true,this,null);if(chainedWatcher!=null)chainedW

50、cess(event);如果客戶端Zookeeper程序在會(huì)話失效時(shí)(Expiredevent)重新建立了通信信道(SyncConnectedevent),所有的會(huì)話監(jiān)視會(huì)自動(dòng)和服務(wù)器進(jìn)行重連，(Zookeeper更多編程指南請(qǐng)參見ZooKeeperWatches。當(dāng)DataMonitor獲得了指定Znode的事件后，它將調(diào)用ZooKeeper.exists()來決定究竟發(fā)生了什么。完整的程序:Executor.java：/*AsimpleexampleprogramtouseDataMonitortostartandstopexecutablesbasedonaznode.Theprogra

51、mwatchesthespecifiedznodeandsavesthedatathatcorrespondstotheznodeinthefilesystem.Italsostartsthespecifiedprogramwiththespecifiedargumentswhentheznodeexistsandkillstheprogramiftheznodegoesaway.*/import;import;import;import;import;import;import;import;publicclassExecutorimplementsWatcher,Runnable,Data

52、Monitor.DataMonitorListenerStringznode;DataMonitordm;ZooKeeperzk;Stringfilename;Stringexec;Processchild;String exec)publicExecutor(StringhostPort,Stringznode,Stringfilename,throwsKeeperException,IOExceptionthis.filename=filename;this.exec=exec;zk=newZooKeeper(hostPort,3000,this);dm=newDataMonitor(zk

53、,znode,null,this);/*paramargs*/publicstaticvoidmain(Stringargs)if(args.length<4)System.err.println("USAGE:ExecutorhostPortznodefilenameprogramargs.");System.exit(2);StringhostPort=args0;Stringznode=args1;Stringfilename=args2;Stringexec=newStringargs.length-3;System.arraycopy(args,3,exec

54、,0,exec.length);trynewExecutor(hostPort,znode,filename,exec).run();catch(Exceptione)e.printStackTrace();/* Wedoprocessanyeventsourselves,wejustneedtoforwardthemon.* see* /publicvoidprocess(WatchedEventevent)cess(event);publicvoidrun()trysynchronized(this)while(!dm.dead)wait();catch(Interrupted

55、Exceptione)publicvoidclosing(intrc)synchronized(this)notifyAll();staticclassStreamWriterextendsThreadOutputStreamos;InputStreamis;StreamWriter(InputStreamis,OutputStreamos)this.is=is;this.os=os;start();publicvoidrun()byteb=newbyte80;intrc;trywhile(rc=is.read(b)>0)os.write(b,0,rc);catch(IOExceptio

56、ne)publicvoidexists(bytedata)if(data=null)if(child!=null)"Killingprocess");child.destroy();trychild.waitFor();catch(InterruptedExceptione)child=null;elseif(child!=null)"Stoppingchild");child.destroy();trychild.waitFor();catch(InterruptedExceptione)e.printStackTrace();tryFileOutputStreamfos=newFileOutputStream(filename);fos.write(data);fos.close();catch(IOExceptione)e.printStackTrace(