Java中间件-zookeeper（Java中间件错误整理）

一、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里面，之后其它的服务就可以使用了

私信666领取资料