由于zookeeper的client只有zookeeper一個(gè)對(duì)象，使用也比較簡(jiǎn)單，所以就不許要文字說(shuō)明了，在代碼中注釋下就ok 了。

1、測(cè)試用的main方法

package ClientExample; public class TestMain { public static void main(String[] args) { /* * 測(cè)試流程 * 1、創(chuàng)建sever1的連接client1，并且創(chuàng)建一個(gè)永久性的/test節(jié)點(diǎn) * 2、創(chuàng)建一個(gè)針對(duì)server1的臨時(shí)節(jié)點(diǎn) * 3、創(chuàng)建server2的連接client21，并創(chuàng)建一個(gè)針對(duì)server2的臨時(shí)節(jié)點(diǎn) * 4、創(chuàng)建server3的連接client3，并創(chuàng)建一個(gè)針對(duì)server3的臨時(shí)節(jié)點(diǎn) * 5、分別查看client1、client2、client3的三個(gè)節(jié)點(diǎn)的字節(jié)點(diǎn)數(shù)量，確定是否同步成功 * 6、修改client1的臨時(shí)節(jié)點(diǎn)內(nèi)容，然后在在client2和client3中查看 * 7、kill掉client3的線程，然后檢查是watcher是否有通知給client1和client2 */ Thread t1= new ClientThread("127.0.0.1:2181","server1",false); Thread t2= new ClientThread("127.0.0.1:2182","server2",false); Thread t3= new ClientThread("127.0.0.1:2183","server3",false); Thread t4= new ClientThread("127.0.0.1:2181","server4",false); t1.start(); t2.start(); t3.start(); t4.start(); ControlThread c = new ControlThread(t1, t2, t3, t4); c.start(); int i=0; while(true) { i++; i--; } /* * 測(cè)試控制臺(tái)輸出： * connectIP:server4,path:null,state:SyncConnected,type:None * connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged * connectIP:server4,path:/test/server4,state:SyncConnected,type:NodeCreated * 。。。。。。。。。。。 * * connectIP:server2,path:null,state:Disconnected,type:None server2exception,KeeperErrorCode = ConnectionLoss for /test connectIP:newServer1,path:null,state:SyncConnected,type:None connectIP:server1,path:/test,state:SyncConnected,type:NodeChildrenChanged connectIP:server4,path:/test/server2,state:SyncConnected,type:NodeDeleted connectIP:server4,path:/test,state:SyncConnected,type:NodeChildrenChanged connectIP:newServer1,path:/test,state:SyncConnected,type:NodeChildrenChanged connectIP:server3,path:/test/server2,state:SyncConnected,type:NodeDeleted connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged */ } }

2、zookeeper封裝的接口：

package ClientExample; import java.io.IOException; import java.util.List; import org.apache.zookeeper.KeeperException; /** * zookeeper的操作封裝接口，實(shí)現(xiàn)了常用的操作 * 創(chuàng)建、銷毀、寫(xiě)入、修改、查詢等。 * @author ransom * */ public interface ServerOperation { void init(String address,String serverName) throws IOException; void destroy() throws InterruptedException; List<String> getChilds(String path) throws KeeperException, InterruptedException; String getData(String path) throws KeeperException, InterruptedException; void changeData(String path, String data) throws KeeperException, InterruptedException; void delData(String path) throws KeeperException, InterruptedException; void apendTempNode(String path, String data) throws KeeperException, InterruptedException; void apendPresistentNode(String path, String data) throws KeeperException, InterruptedException; void delNode(String path) throws KeeperException, InterruptedException; boolean exist(String path) throws KeeperException, InterruptedException; }

3、接口的實(shí)現(xiàn)：
package ClientExample;import java.io.IOException; import java.util.List; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.ZooDefs.Ids; public class ServerConnector implements ServerOperation { // 創(chuàng)建一個(gè)Zookeeper實(shí)例，第一個(gè)參數(shù)為目標(biāo)服務(wù)器地址和端口，第二個(gè)參數(shù)為Session超時(shí)時(shí)間，第三個(gè)為節(jié)點(diǎn)變化時(shí)的回調(diào)方法 private ZooKeeper zk = null; public void init(String address,String serverName) throws IOException { zk = new ZooKeeper(address, 500000, new MultiWatcher(serverName)); } @Override public void destroy() throws InterruptedException { // TODO Auto-generated method stub if (zk != null) { zk.close(); } } @Override public List<String> getChilds(String path) throws KeeperException, InterruptedException { // TODO Auto-generated method stub if (zk != null) { return zk.getChildren(path, true); } return null; } @Override public String getData(String path) throws KeeperException, InterruptedException { // TODO Auto-generated method stub if (zk != null) { // 取得/root/childone節(jié)點(diǎn)下的數(shù)據(jù),返回byte[] byte[] b = zk.getData(path, true, null); return new String(b); } return null; } @Override public void changeData(String path,String data) throws KeeperException, InterruptedException { // TODO Auto-generated method stub if (zk != null) { // 修改節(jié)點(diǎn)/root/childone下的數(shù)據(jù)，第三個(gè)參數(shù)為版本，如果是-1，那會(huì)無(wú)視被修改的數(shù)據(jù)版本，直接改掉 zk.setData(path, data.getBytes(),-1); } } @Override public void delData(String path) throws InterruptedException, KeeperException { // TODO Auto-generated method stub if (zk != null) { // 刪除/root/childone這個(gè)節(jié)點(diǎn)，第二個(gè)參數(shù)為版本，－1的話直接刪除，無(wú)視版本 zk.delete(path, -1); } } @Override public void delNode(String path) throws InterruptedException, KeeperException { // TODO Auto-generated method stub if (zk != null) { zk.delete(path, -1); } } @Override public boolean exist(String path) throws KeeperException, InterruptedException { // TODO Auto-generated method stub if (zk != null) { return zk.exists(path, true)!=null; } return false; } @Override public void apendTempNode(String path, String data) throws KeeperException, InterruptedException { // TODO Auto-generated method stub // TODO Auto-generated method stub if (zk != null) { // 創(chuàng)建一個(gè)節(jié)點(diǎn)root，數(shù)據(jù)是mydata,不進(jìn)行ACL權(quán)限控制，節(jié)點(diǎn)為永久性的(即客戶端shutdown了也不會(huì)消失) /* * 創(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ǔ)的數(shù)據(jù)不會(huì)丟失； * PERSISTENT_SEQUENTIAL：順序自動(dòng)編號(hào)的目錄節(jié)點(diǎn)，這種目錄節(jié)點(diǎn)會(huì)根據(jù)當(dāng)前已近存在的節(jié)點(diǎn)數(shù)自動(dòng)加 1，然后返回給客戶端已經(jīng)成功創(chuàng)建的目錄節(jié)點(diǎn)名； * EPHEMERAL：臨時(shí)目錄節(jié)點(diǎn)，一旦創(chuàng)建這個(gè)節(jié)點(diǎn)的客戶端與服務(wù)器端口也就是 session 超時(shí)，這種節(jié)點(diǎn)會(huì)被自動(dòng)刪除； * EPHEMERAL_SEQUENTIAL：臨時(shí)自動(dòng)編號(hào)節(jié)點(diǎn) */ zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); } } @Override public void apendPresistentNode(String path, String data) throws KeeperException, InterruptedException { // TODO Auto-generated method stub if (zk != null) { // 創(chuàng)建一個(gè)節(jié)點(diǎn)root，數(shù)據(jù)是mydata,不進(jìn)行ACL權(quán)限控制，節(jié)點(diǎn)為永久性的(即客戶端shutdown了也不會(huì)消失) /* * 創(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ǔ)的數(shù)據(jù)不會(huì)丟失； * PERSISTENT_SEQUENTIAL：順序自動(dòng)編號(hào)的目錄節(jié)點(diǎn)，這種目錄節(jié)點(diǎn)會(huì)根據(jù)當(dāng)前已近存在的節(jié)點(diǎn)數(shù)自動(dòng)加 1，然后返回給客戶端已經(jīng)成功創(chuàng)建的目錄節(jié)點(diǎn)名； * EPHEMERAL：臨時(shí)目錄節(jié)點(diǎn)，一旦創(chuàng)建這個(gè)節(jié)點(diǎn)的客戶端與服務(wù)器端口也就是 session 超時(shí)，這種節(jié)點(diǎn)會(huì)被自動(dòng)刪除； * EPHEMERAL_SEQUENTIAL：臨時(shí)自動(dòng)編號(hào)節(jié)點(diǎn) */ zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } } 4、一個(gè)控制的線程，主要用來(lái)強(qiáng)制kill掉連接的線程

package ClientExample; public class ControlThread extends Thread{ public ControlThread(Thread t1,Thread t2,Thread t3,Thread t4) { list[0]=t1; list[1]=t2; list[2]=t4; list[3]=t4; } private Thread[] list = new Thread[4]; private int num=0; public void run() { while(true) { if(num==7) { list[2].stop(); System.out.println("kill server3"); } if(num==15) { list[3].stop(); System.out.println("kill server4"); } try { sleep(1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } }

5、watcher 的實(shí)現(xiàn)：

package ClientExample; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.Watcher.Event.EventType; import org.apache.zookeeper.Watcher.Event.KeeperState; /** * 提供給多個(gè)client使用的watcher * @author ransom * */ public class MultiWatcher implements Watcher{ public MultiWatcher(String address) { connectAddress=address; } private String connectAddress=null; @Override public void process(WatchedEvent event) { // TODO Auto-generated method stub String outputStr=""; if(connectAddress!=null){ outputStr+="connectIP:"+connectAddress; } outputStr+=",path:"+event.getPath(); outputStr+=",state:"+event.getState(); outputStr+=",type:"+event.getType(); System.out.println(outputStr); } }

6、client 運(yùn)行 的Thread

package ClientExample; import java.io.IOException; import java.text.DateFormat; import java.text.SimpleDateFormat; import java.util.Date; import java.util.List; import org.apache.zookeeper.KeeperException; public class ClientThread extends Thread{ public ClientThread(String address,String serverName,boolean islog) { this.address=address; this.serverName=serverName; try { otherOperation(); } catch (KeeperException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } this.islog=islog; } private boolean islog=true; private final String rootPath = "/test"; private String address; private String serverName; private ServerOperation operationCient = null; public void run() { if(operationCient==null) { System.out.println("operationCient=null"); return; } while(true){ try { if(islog){ System.out.println(serverName+",loopTime:"+getNowTime()); } observerChildData(rootPath); } catch (KeeperException e) { // TODO Auto-generated catch block System.out.println(serverName+"exception,"+e.getLocalizedMessage()); try { operationCient= new ServerConnector(); operationCient.init("127.0.0.1:2181","newServer1"); } catch (IOException e1) { // TODO Auto-generated catch block System.out.println(serverName+" reconnect exception,"+e.getLocalizedMessage()); } } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } try { sleep(2000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } /* * 測(cè)試流程 * 1、創(chuàng)建sever1的連接client1，并且創(chuàng)建一個(gè)永久性的/test節(jié)點(diǎn) * 2、創(chuàng)建一個(gè)針對(duì)server1的臨時(shí)節(jié)點(diǎn) * 3、創(chuàng)建server2的連接client21，并創(chuàng)建一個(gè)針對(duì)server2的臨時(shí)節(jié)點(diǎn) * 4、創(chuàng)建server3的連接client3，并創(chuàng)建一個(gè)針對(duì)server3的臨時(shí)節(jié)點(diǎn) * 5、分別查看client1、client2、client3的三個(gè)節(jié)點(diǎn)的字節(jié)點(diǎn)數(shù)量，確定是否同步成功 * 6、修改client1的臨時(shí)節(jié)點(diǎn)內(nèi)容，然后在在client2和client3中查看 * 7、kill掉client3的線程，然后檢查是watcher是否有通知給client1和client2 */ private void otherOperation() throws KeeperException, InterruptedException { operationCient= new ServerConnector(); try { operationCient.init(address,serverName); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } if(operationCient==null) { System.out.println("operationCient=null"); return; } if(!operationCient.exist(rootPath)) { operationCient.apendPresistentNode(rootPath, "this node is creat by " + serverName); } //添加臨時(shí)節(jié)點(diǎn) if(!operationCient.exist(rootPath+"/"+serverName)) { operationCient.apendTempNode(rootPath+"/"+serverName, "this node is creat by " + serverName); } observerChildData("/test"); //修改臨時(shí)節(jié)點(diǎn)內(nèi)容 operationCient.changeData(rootPath+"/"+serverName, "this node is changed by " + serverName); //臨時(shí)節(jié)點(diǎn)內(nèi)容 List<String> childs=operationCient.getChilds(rootPath); for(String str : childs) { System.out.println("observered by "+ serverName +": child node is :"+ str); } } //查看臨時(shí)節(jié)點(diǎn)的同步狀態(tài) public void observerChildData(String path) throws KeeperException, InterruptedException { if(operationCient==null) { System.out.println("operationCient=null"); return; } List<String> childs=operationCient.getChilds(rootPath); if(islog){ System.out.println("observered by "+ serverName +": childs len is :"+ childs.size()); } for(String str : childs) { if(islog){ System.out.println("observered by "+ serverName +": child node is :"+ str+",data is :"+operationCient.getData(rootPath+"/"+str)); } } } public String getNowTime() { DateFormat format1 = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss"); return format1.format(new Date()); } }

轉(zhuǎn)載于:https://www.cnblogs.com/cl1024cl/p/6205140.html

總結(jié)

以上是生活随笔為你收集整理的zookeeper 应用开发的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。