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

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

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

4、每個serverOO#文件夾下面新建data和logs子文件夾。Zookeeper的配置文件主要在conf目錄，包括zoo.cfg（zoo_sample.cfg）和perties,修改zoo_sample.cfg,重命名為zoo.cgf,打開zoo.cfg,內容如下：# 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將內容修改為（server001節點的配置文件）：# 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：這個時間是作為Zookeeper服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每個tickTime時間就會發送一個心跳。?dataDir：顧名思義就是Zookeeper保存數據的目錄，默認情況下，Zookeeper將寫數據的日志文件也保存在這個目錄里。?clientPort：這個端口就是客戶端連接Zookeeper服務器的端口，Zookeeper會監

10、聽這個端口，接受客戶端的訪問請求。?initLimit：這個配置項是用來配置Zookeeper接受客戶端（這里所說的客戶端不是用戶連接Zookeeper服務器的客戶端，而是Zookeeper服務器集群中連接到Leader的Follower服務器）初始化連接時最長能忍受多少個心跳時間間隔數。當已經超過5個心跳的時間（也就是tickTime）長度后Zookeeper服務器還沒有收到客戶端的返回信息，那么表明這個客戶端連接失敗。總的時間長度就是5*2000=10秒?syncLimit：這個配置項標識Leader與Follower之間發送消息，請求和應答時間長度，最長不能超過多少個tickTime的時

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

12、r/server001/data/下建立一個myid文件，文件內容為1，echo"1">>/tmp/zookeeper/server001/data/myid繼續修改/zookeeper/目錄中的zookeeper配置文件文件（server002的配置文件，注意clientPort=2182，與server001中的clientPort=2181不同，后續修改配置均需設置不同的clientPort），內容如下：# 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/下建立一個myid文件，文件內容為2，echo"2">>/tmp/zookeeper/server001/data/myid依次修改配置文件，建立server003，server004，server005節點文件夾，

16、完成上述步驟后/tmp/zookeeper目錄結構如下：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然后依次進入每個文件夾節點的zookeeper目錄中，啟動zookee

17、per服務，$bin/zkServer.shstart如果一切順利，Zookeeper偽分布式模式安裝成功，下面驗證Zookeeper安裝的正確性。進入任意一個文件夾節點的zookeeper包所在的目錄，執行一下命令：$bin/zkCli.sh-server執行成功后：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典型的應用場景。簡介ApacheZookeeper是由ApacheHadoop的Zookeeper子項目發展而來，現在已經成為了Apache的頂級項目。Zookeeper為分布式系統提供了高效可靠且易于使用的協同服務，它可以為分布式應用提供相當多的服務，諸如統一命名服務，配置管理，狀態同步和組服務等。Zookeeper接口簡單，開發人員不必過多地結結在分布式系統編程難于處理的同步和一致性問題上，你可以使用Zookeeper提供的現成(off-the

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

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

22、話有效。只要客戶端會話有效，應用程序可以調用Zookeeper客戶端的接口與服務器端進行交互。下表主要介紹了ZooKeeper方法(該表摘自IBMDeveloperworks:分布式服務框架Zookeeper-管理分布式環境中的數據)方法名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)方法功能描述創建一個給定的目錄節點path,并給它設置數據，CreateMode標識有四種形式的目錄節點，分別是PERSISTENT持久化目錄節點，這個目錄節點存儲不會丟失;PERSISTENT_SEQUENT:IAL序自動編三節點，這種目錄節點會根據當前已近存在的節點數1,然后返回給客戶端已經成功創建的目錄節點名；EPHEMERAL臨

24、時目錄節點,一旦創建這個節點的客服務器端口也就是session超時，這種節點會被自除；EPHEMERAL_SEQUENTIAW自動編號節點。判斷某個path是否存在，并設置是否監控這個目這里的watcher是在創建ZooKeeper實例時指定watcher,exists方法還有一個重載方法,可以指Iwatcher。重載方法，這里給某個目錄節點設置特定的watcWatcher在ZooKeeper是一個核心功能，Watche控目錄節點的數據變化以及子目錄的變化，一旦這發生變化，服務器就會通知所有設置在這個目錄節Watcher,從而每個客戶端都很快知道它所關注的目的狀態發生變化，而做出相應的反應。刪

25、除path對應的目錄節點，version為-1可力何版本，也就刪除了這個目錄節點所有數據。獲取指定path下的所有子目錄節點，同樣getCI方法也有一個重載方法可以設置特定的watcher.節點的狀態。給path設置數據，可以指定這個數據的版本號，version為-1怎可以匹配任何版本。byte口getData(Stringpath,booleanwatch,獲取這個path對應的目錄節點存儲的數據，數據StatstOt)等信息可以通過stat來指定，同時還可以設置是一這個目錄節點數據的狀態。voidaddAuthInfo(Stringscheme,byte口auth)客戶端將自己的授權信息提

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

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

28、ewWatcher()/監控所有被觸勺事件publicvoidprocess(WatchedEventevent)"已經觸發了"+event.getType()+"事件！");)；/創建一個目錄節點zk.create("/testRootPath","testRootData".getBytes(),Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);/創建一個子目錄節點zk.create("/testRootPath/testChildPathOne",&qu

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

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

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

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

33、止某一程序。基本要求該客戶端具備四個基本要求：?客戶端所帶參數：oZookeeper服務地址。o被監視的Znode節點名稱。o可執行程序及其所帶的參數?客戶端會獲取被監視Znode節點的數據并啟動你所指定的可執行程序。?如果被監視的Znode節點發生改變，客戶端重新獲取其內容并再次啟動你所指定的可執行程序。?如果被監視的Znode節點消失，客戶端會殺死可執行程序。程序設計一般而言，Zookeeper應用程序分為兩部分，其中一部分維護與服務器端的連接，另外一部分監視Znode節點的數據。在本程序中，Executor類負責維護Zookeeper連接，DataMonitor類監視Zookeeper目

34、錄樹中的數據，同時，Executor包含了主線程和程序主要的執行邏輯，它負責少量的用戶交互，以及與可執行程序的交互，該可執行程序接受你向它傳入的參數，并且會根據被監視的Znode節點的狀態變化停止或重啟。Executor類Executor對象是本例程最基本的“容器”，它包括Zookeeper對象和DataMonitor對象。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的任務是根據Zookeeper中Znode節點狀態改變所觸發的事件來啟動和停止你在命令行指定的可執行程序，

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

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

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

40、voidclosing(intrc);該接口在DataMonitor中定義，Executor類實現該接口，當Executor.exists()被調用的時候，Executor決定是否啟動或停止事先指定的應用程序(回憶一下前文所說的，當Znode消失時Zookeeper客戶端會殺死該可執行程序)。當Executor.closing()被調用的時候，Executor會根據Zookeeper連接永久性地消失來決定是否關閉自己。你或許已經猜到，DataMonitor對象根據Zookeeper狀態變化來調用這些方法吧？以下是Executor類中實現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邏輯的核心，它差不多是異步的，并由事件驅動的。DataMonitor構造函數如下：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);調用ZooKeeper.exists()檢查指定的Znode是否存在，并設置監視，傳遞自身引用作為回調對象，在某種意義上，在watch觸發時就會引起真實的處理流程。當ZooKeeper.exists()操作在服務器端完成時，ZooKeeperAPI會在客戶端調用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是否存在，以及其他重大的不可恢復的錯誤。如果文件(或者Znode)存在，它將從Znode獲取數據，如果狀態發生變化再調用Executor的exists()回調函數。注意，getData函數本省必須要做任何的異常處理，因為本身就有監視可以處理任何錯誤：如果節點在調用ZooKeeper.getData()之前被刪除，ZooKeeper.exists()就會觸發回調函數，如果存在通信錯誤，在連接上的監視會在該連接重建之前觸發相應的事件，同時引發相應的處理。最后，DataMonitor處理監視事件的代碼如下：

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程序在會話失效時(Expiredevent)重新建立了通信信道(SyncConnectedevent),所有的會話監視會自動和服務器進行重連，(Zookeeper更多編程指南請參見ZooKeeperWatches。當DataMonitor獲得了指定Znode的事件后，它將調用ZooKeeper.exists()來決定究竟發生了什么。完整的程序: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(