快捷導(dǎo)航

java 中 zookeeper簡單使用

更新時(shí)間：2017年09月27日 10:30:34 投稿：mrr

ZooKeeper是一個(gè)分布式的，開放源碼的分布式應(yīng)用程序協(xié)調(diào)服務(wù)，是Google的Chubby一個(gè)開源的實(shí)現(xiàn)，是Hadoop和Hbase的重要組件。下面通過本文給大家分享java 中 zookeeper簡單使用，需要的朋友參考下吧

一、zookeeper的基本原理

數(shù)據(jù)模型，如下：

ZooKeeper數(shù)據(jù)模型的結(jié)構(gòu)與Unix文件系統(tǒng)很類似，整體上可以看作是一棵樹，每個(gè)節(jié)點(diǎn)稱做一個(gè)ZNode。每個(gè)ZNode都可以通過其路徑唯一標(biāo)識(shí)，比如上圖中第三層的第一個(gè)ZNode,它的路徑是/app1/c1。在每個(gè)ZNode上可存儲(chǔ)少量數(shù)據(jù)(默認(rèn)是1M, 可以通過配置修改,通常不建議在ZNode上存儲(chǔ)大量的數(shù)據(jù))，這個(gè)特性非常有用。另外，每個(gè)ZNode上還存儲(chǔ)了其Acl信息，這里需要注意，雖說ZNode的樹形結(jié)構(gòu)跟Unix文件系統(tǒng)很類似，但是其Acl與Unix文件系統(tǒng)是完全不同的，每個(gè)ZNode的Acl的獨(dú)立的，子結(jié)點(diǎn)不會(huì)繼承父結(jié)點(diǎn)的。

ZooKeeper特性：

1、讀、寫(更新)模式

在ZooKeeper集群中，讀可以從任意一個(gè)ZooKeeperServer讀，這一點(diǎn)是保證ZooKeeper比較好的讀性能的關(guān)鍵；寫的請(qǐng)求會(huì)先Forwarder到Leader，然后由Leader來通過ZooKeeper中的原子廣播協(xié)議，將請(qǐng)求廣播給所有的Follower，Leader收到一半以上的寫成功的Ack后，就認(rèn)為該寫成功了，就會(huì)將該寫進(jìn)行持久化，并告訴客戶端寫成功了。

2、WAL和Snapshot

和大多數(shù)分布式系統(tǒng)一樣，ZooKeeper也有WAL(Write-Ahead-Log)，對(duì)于每一個(gè)更新操作，ZooKeeper都會(huì)先寫WAL,然后再對(duì)內(nèi)存中的數(shù)據(jù)做更新，然后向Client通知更新結(jié)果。另外，ZooKeeper還會(huì)定期將內(nèi)存中的目錄樹進(jìn)行Snapshot，落地到磁盤上，這個(gè)跟HDFS中的FSImage是比較類似的。這么做的主要目的，一當(dāng)然是數(shù)據(jù)的持久化，二是加快重啟之后的恢復(fù)速度，如果全部通過ReplayWAL的形式恢復(fù)的話，會(huì)比較慢。

3、FIFO

對(duì)于每一個(gè)ZooKeeper客戶端而言，所有的操作都是遵循FIFO順序的，這一特性是由下面兩個(gè)基本特性來保證的：一是ZooKeeperClient與Server之間的網(wǎng)絡(luò)通信是基于TCP，TCP保證了Client/Server之間傳輸包的順序；二是ZooKeeperServer執(zhí)行客戶端請(qǐng)求也是嚴(yán)格按照FIFO順序的。

4、Linearizability

在ZooKeeper中，所有的更新操作都有嚴(yán)格的偏序關(guān)系，更新操作都是串行執(zhí)行的，這一點(diǎn)是保證ZooKeeper功能正確性的關(guān)鍵。

二、zookeeper的常用命令

我們可以執(zhí)行zookeeper-client或者執(zhí)行/opt/cloudera/parcels/CDH-5.0.0-1.cdh5.0.0.p0.47/lib/zookeeper/bin/zkCli.sh-server localhost，進(jìn)入zookeeper命令行，如下：

然后，執(zhí)行l(wèi)s /可以看到：

然后，我們可以執(zhí)行create /qyktest‘qyktest'創(chuàng)建一個(gè)節(jié)點(diǎn)，如下：

然后，我們執(zhí)行g(shù)et /qyktest獲取節(jié)點(diǎn)值，如下：

然后，我們可以執(zhí)行set /qyktest‘111'修改節(jié)點(diǎn)的值，如下：

最后，我們執(zhí)行delete /qyktest便可刪除此節(jié)點(diǎn)。

另外，我們還可以在qyktest此節(jié)點(diǎn)下繼續(xù)創(chuàng)建子節(jié)點(diǎn)。

好了，幾個(gè)基本命令就講到這人啦，其它的命令還有很多，大家可以去查閱下資料。

三、zookeeper的javaapi操作

關(guān)于Javaapi操作zookeeper比較簡單，筆者直接貼出代碼，如下：

packageorg.zookeeper.demo;
importjava.io.IOException;
importjava.util.concurrent.CountDownLatch;
importorg.apache.zookeeper.CreateMode;
importorg.apache.zookeeper.KeeperException;
importorg.apache.zookeeper.WatchedEvent;
importorg.apache.zookeeper.Watcher;
importorg.apache.zookeeper.Watcher.Event.KeeperState;
importorg.apache.zookeeper.ZooDefs.Ids;
importorg.apache.zookeeper.ZooKeeper;
publicclassZookeeperClientimplementsWatcher{
//連接超時(shí)時(shí)間,10s
privatestaticfinalintSESSION_TIMEOUT= 10000;
//連接的zookeeperserver
privatestaticfinalStringCONNECTION_STRING = "172.31.25.8:2181";
privatestaticfinalStringZK_PATH = "/qyktest";
privateZooKeeperzk = null;
privateCountDownLatchconnectedSemaphore = newCountDownLatch(1);
publicvoidcreateConnection(StringconnectString, intsessionTimeout){
this.releaseConnection();
try{
zk= newZooKeeper(connectString,sessionTimeout, this);
connectedSemaphore.await();
}catch(InterruptedExceptione) {
System.out.println("連接創(chuàng)建失敗，發(fā)生InterruptedException");
e.printStackTrace();
}catch(IOExceptione) {
System.out.println("連接創(chuàng)建失敗，發(fā)生IOException");
e.printStackTrace();
}
}
publicvoidreleaseConnection(){
if(this.zk!= null){
try{
this.zk.close();
}catch(InterruptedExceptione) {
e.printStackTrace();
}
}
}
publicbooleancreatePath(Stringpath, String data) {
try{
Stringresult = this.zk.create(path,data.getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
System.out.println("節(jié)點(diǎn)創(chuàng)建成功,Path: "+result + ", content: "+data);
}catch(KeeperExceptione) {
System.out.println("節(jié)點(diǎn)創(chuàng)建失敗，發(fā)生KeeperException");
e.printStackTrace();
}catch(InterruptedExceptione) {
System.out.println("節(jié)點(diǎn)創(chuàng)建失敗，發(fā)生InterruptedException");
e.printStackTrace();
}
returntrue;
}
publicStringreadData(Stringpath) {
try{
System.out.println("獲取數(shù)據(jù)成功，path："+path);
returnnewString(this.zk.getData(path,false,null));
}catch(KeeperExceptione) {
System.out.println("讀取數(shù)據(jù)失敗，發(fā)生KeeperException，path:"+path);
e.printStackTrace();
return"";
}catch(InterruptedExceptione) {
System.out.println("讀取數(shù)據(jù)失敗，發(fā)生InterruptedException，path: "+path);
e.printStackTrace();
return"";
}
}
publicbooleanwriteData(Stringpath, String data) {
try{
System.out.println("更新數(shù)據(jù)成功，path："+path + ", stat: "+this.zk.setData(path,data.getBytes(), -1));
}catch(KeeperExceptione) {
System.out.println("更新數(shù)據(jù)失敗，發(fā)生KeeperException，path:"+path);
e.printStackTrace();
}catch(InterruptedExceptione) {
System.out.println("更新數(shù)據(jù)失敗，發(fā)生InterruptedException，path: "+path);
e.printStackTrace();
}
returnfalse;
}
publicvoiddeleteNode(Stringpath) {
try{
this.zk.delete(path,-1);
System.out.println("刪除節(jié)點(diǎn)成功，path："+path);
}catch(KeeperExceptione) {
System.out.println("刪除節(jié)點(diǎn)失敗，發(fā)生KeeperException，path:"+path);
e.printStackTrace();
}catch(InterruptedExceptione) {
System.out.println("刪除節(jié)點(diǎn)失敗，發(fā)生InterruptedException，path: "+path);
e.printStackTrace();
}
}
publicstaticvoidmain(String[]args) {
ZookeeperClientsample = newZookeeperClient();
//獲取連接
sample.createConnection(CONNECTION_STRING,SESSION_TIMEOUT);
//讀數(shù)據(jù)
Stringqyk = sample.readData("/qyktest");
System.out.println("qyk:"+qyk);
Stringurl = sample.readData("/qyk/db/url");
System.out.println("url"+url);
Stringdriver = sample.readData("/qyk/db/driver");
System.out.println("driver"+driver);
StringuserName = sample.readData("/qyk/db/userName");
System.out.println("userName"+userName);
Stringpassword = sample.readData("/qyk/db/password");
System.out.println("password"+password);
//創(chuàng)建節(jié)點(diǎn)
sample.createPath(ZK_PATH,"我是節(jié)點(diǎn)初始內(nèi)容");
System.out.println("數(shù)據(jù)內(nèi)容:"+sample.readData(ZK_PATH) + "\n");
//更新節(jié)點(diǎn)
sample.writeData(ZK_PATH,"更新后的數(shù)據(jù)");
System.out.println("數(shù)據(jù)內(nèi)容:"+sample.readData(ZK_PATH) + "\n");
//刪除節(jié)點(diǎn)
sample.deleteNode(ZK_PATH);
//釋放連接
sample.releaseConnection();
}
@Override
publicvoidprocess(WatchedEventevent) {
System.out.println("收到事件通知："+event.getState() + "\n");
if(KeeperState.SyncConnected== event.getState()) {
connectedSemaphore.countDown();
}
}
}

然后，執(zhí)行可以看到，控制臺(tái)輸出如下：