java 中 zookeeper简单使用
一、zookeeper的基本原理
数据模型,如下:
ZooKeeper数据模型的结构与Unix文件系统很类似,整体上可以看作是一棵树,每个节点称做一个ZNode。每个ZNode都可以通过其路径唯一标识,比如上图中第三层的第一个ZNode,它的路径是/app1/c1。在每个ZNode上可存储少量数据(默认是1M, 可以通过配置修改,通常不建议在ZNode上存储大量的数据),这个特性非常有用。另外,每个ZNode上还存储了其Acl信息,这里需要注意,虽说ZNode的树形结构跟Unix文件系统很类似,但是其Acl与Unix文件系统是完全不同的,每个ZNode的Acl的独立的,子结点不会继承父结点的。
ZooKeeper特性:
1、读、写(更新)模式
在ZooKeeper集群中,读可以从任意一个ZooKeeperServer读,这一点是保证ZooKeeper比较好的读性能的关键;写的请求会先Forwarder到Leader,然后由Leader来通过ZooKeeper中的原子广播协议,将请求广播给所有的Follower,Leader收到一半以上的写成功的Ack后,就认为该写成功了,就会将该写进行持久化,并告诉客户端写成功了。
2、WAL和Snapshot
和大多数分布式系统一样,ZooKeeper也有WAL(Write-Ahead-Log),对于每一个更新操作,ZooKeeper都会先写WAL,然后再对内存中的数据做更新,然后向Client通知更新结果。另外,ZooKeeper还会定期将内存中的目录树进行Snapshot,落地到磁盘上,这个跟HDFS中的FSImage是比较类似的。这么做的主要目的,一当然是数据的持久化,二是加快重启之后的恢复速度,如果全部通过ReplayWAL的形式恢复的话,会比较慢。
3、FIFO
对于每一个ZooKeeper客户端而言,所有的操作都是遵循FIFO顺序的,这一特性是由下面两个基本特性来保证的:一是ZooKeeperClient与Server之间的网络通信是基于TCP,TCP保证了Client/Server之间传输包的顺序;二是ZooKeeperServer执行客户端请求也是严格按照FIFO顺序的。
4、Linearizability
在ZooKeeper中,所有的更新操作都有严格的偏序关系,更新操作都是串行执行的,这一点是保证ZooKeeper功能正确性的关键。
二、zookeeper的常用命令
我们可以执行zookeeper-client或者执行/opt/cloudera/parcels/CDH-5.0.0-1.cdh5.0.0.p0.47/lib/zookeeper/bin/zkCli.sh-server localhost,进入zookeeper命令行,如下:
然后,执行ls /可以看到:
然后,我们可以执行create /qyktest‘qyktest'创建一个节点,如下:
然后,我们执行get /qyktest获取节点值,如下:
然后,我们可以执行set /qyktest‘111'修改节点的值,如下:
最后,我们执行delete /qyktest便可删除此节点。
另外,我们还可以在qyktest此节点下继续创建子节点。
好了,几个基本命令就讲到这人啦,其它的命令还有很多,大家可以去查阅下资料。
三、zookeeper的javaapi操作
关于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{ //连接超时时间,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("连接创建失败,发生InterruptedException"); e.printStackTrace(); }catch(IOExceptione) { System.out.println("连接创建失败,发生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("节点创建成功,Path: "+result + ", content: "+data); }catch(KeeperExceptione) { System.out.println("节点创建失败,发生KeeperException"); e.printStackTrace(); }catch(InterruptedExceptione) { System.out.println("节点创建失败,发生InterruptedException"); e.printStackTrace(); } returntrue; } publicStringreadData(Stringpath) { try{ System.out.println("获取数据成功,path:"+path); returnnewString(this.zk.getData(path,false,null)); }catch(KeeperExceptione) { System.out.println("读取数据失败,发生KeeperException,path:"+path); e.printStackTrace(); return""; }catch(InterruptedExceptione) { System.out.println("读取数据失败,发生InterruptedException,path: "+path); e.printStackTrace(); return""; } } publicbooleanwriteData(Stringpath, String data) { try{ System.out.println("更新数据成功,path:"+path + ", stat: "+this.zk.setData(path,data.getBytes(), -1)); }catch(KeeperExceptione) { System.out.println("更新数据失败,发生KeeperException,path:"+path); e.printStackTrace(); }catch(InterruptedExceptione) { System.out.println("更新数据失败,发生InterruptedException,path: "+path); e.printStackTrace(); } returnfalse; } publicvoiddeleteNode(Stringpath) { try{ this.zk.delete(path,-1); System.out.println("删除节点成功,path:"+path); }catch(KeeperExceptione) { System.out.println("删除节点失败,发生KeeperException,path:"+path); e.printStackTrace(); }catch(InterruptedExceptione) { System.out.println("删除节点失败,发生InterruptedException,path: "+path); e.printStackTrace(); } } publicstaticvoidmain(String[]args) { ZookeeperClientsample = newZookeeperClient(); //获取连接 sample.createConnection(CONNECTION_STRING,SESSION_TIMEOUT); //读数据 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); //创建节点 sample.createPath(ZK_PATH,"我是节点初始内容"); System.out.println("数据内容:"+sample.readData(ZK_PATH) + "\n"); //更新节点 sample.writeData(ZK_PATH,"更新后的数据"); System.out.println("数据内容:"+sample.readData(ZK_PATH) + "\n"); //删除节点 sample.deleteNode(ZK_PATH); //释放连接 sample.releaseConnection(); } @Override publicvoidprocess(WatchedEventevent) { System.out.println("收到事件通知:"+event.getState() + "\n"); if(KeeperState.SyncConnected== event.getState()) { connectedSemaphore.countDown(); } } }
然后,执行可以看到,控制台输出如下:
所以,像一些公用的配置,我们可以存到zookeeper里面,之后其它的服务就可以使用了
总结
以上所述是小编给大家介绍的java 中 zookeeper简单使用,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对我们网站的支持!