Java实现与ZooKeeper的连接

細細.魚

序言
　　续上一篇的 ZooKeeper安装运行  这里采用ZooKeeper的Java API进行连接。
　　
Java实现
　　新建一个类实现接口Watcher. 是指：
This interface specifies the public interface an event handler class must implement. A ZooKeeper client will get various events from the ZooKeepr server it connects to. An application using such a client handles these events by registering a callback object with the client. The callback object is expected to be an instance of a class that implements Watcher interface.

/**
* AbstractZooKeeper.java
* 版权所有(C) 2013
* 创建:cuiran 2013-01-16 14:59:44
*/
package com.zoo.demo;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;

/**
* TODO
* @author cuiran
* @version TODO
*/
public class AbstractZooKeeper implements Watcher {
private static Log log = LogFactory.getLog(AbstractZooKeeper.class.getName());
//缓存时间
private static final int SESSION_TIME   = 2000;   
protected ZooKeeper zooKeeper;
protected CountDownLatch countDownLatch=new CountDownLatch(1);
public void connect(String hosts) throws IOException, InterruptedException{   
zooKeeper = new ZooKeeper(hosts,SESSION_TIME,this);   
countDownLatch.await();   
}   
/* (non-Javadoc)
* @see org.apache.zookeeper.Watcher#process(org.apache.zookeeper.WatchedEvent)
*/
@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
if(event.getState()==KeeperState.SyncConnected){
countDownLatch.countDown();
}
}
public void close() throws InterruptedException{   
zooKeeper.close();   
}  
}
然后写一些操作和main方法
/**
* ZooKeeperOperator.java
* 版权所有(C) 2013
* 创建:cuiran 2013-01-16 15:03:40
*/
package com.zoo.demo;
import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
/**
* TODO
* @author cuiran
* @version TODO
*/
public class ZooKeeperOperator extends AbstractZooKeeper {
private static Log log = LogFactory.getLog(ZooKeeperOperator.class.getName());
/**
*
*<b>function:</b>创建持久态的znode,比支持多层创建.比如在创建/parent/child的情况下,无/parent.无法通过
*@author cuiran
*@createDate 2013-01-16 15:08:38
*@param path
*@param data
*@throws KeeperException
*@throws InterruptedException
*/
public void create(String path,byte[] data)throws KeeperException, InterruptedException{
/**
* 此处采用的是CreateMode是PERSISTENT  表示The znode will not be automatically deleted upon client's disconnect.
* EPHEMERAL 表示The znode will be deleted upon the client's disconnect.
*/
this.zooKeeper.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
/**
*
*<b>function:</b>获取节点信息
*@author cuiran
*@createDate 2013-01-16 15:17:22
*@param path
*@throws KeeperException
*@throws InterruptedException
*/
public void getChild(String path) throws KeeperException, InterruptedException{   
try{
List<String> list=this.zooKeeper.getChildren(path, false);
if(list.isEmpty()){
log.debug(path+&quot;中没有节点&quot;);
}else{
log.debug(path+&quot;中存在节点&quot;);
for(String child:list){
log.debug(&quot;节点为：&quot;+child);
}
}
}catch (KeeperException.NoNodeException e) {
// TODO: handle exception
throw e;   
}
}
public byte[] getData(String path) throws KeeperException, InterruptedException {   
return  this.zooKeeper.getData(path, false,null);   
}  
public static void main(String[] args) {
try {   
ZooKeeperOperator zkoperator             = new ZooKeeperOperator();   
zkoperator.connect(&quot;192.168.0.138&quot;);
byte[] data = new byte[]{'a','b','c','d'};   
//            zkoperator.create(&quot;/root&quot;,null);   
//            System.out.println(Arrays.toString(zkoperator.getData(&quot;/root&quot;)));   
//               
//            zkoperator.create(&quot;/root/child1&quot;,data);   
//            System.out.println(Arrays.toString(zkoperator.getData(&quot;/root/child1&quot;)));   
//               
//            zkoperator.create(&quot;/root/child2&quot;,data);   
//            System.out.println(Arrays.toString(zkoperator.getData(&quot;/root/child2&quot;)));   
String zktest=&quot;ZooKeeper的Java API测试&quot;;
zkoperator.create(&quot;/root/child3&quot;, zktest.getBytes());
log.debug(&quot;获取设置的信息：&quot;+new String(zkoperator.getData(&quot;/root/child3&quot;)));
System.out.println(&quot;节点孩子信息:&quot;);   
zkoperator.getChild(&quot;/root&quot;);   
zkoperator.close();   

} catch (Exception e) {   
e.printStackTrace();   
}   
}
}

最后运行结果：
2013-01-16 15:26:04:INFO org.apache.zookeeper.ZooKeeper - Client environment:user.name=Administrator
2013-01-16 15:26:04:INFO org.apache.zookeeper.ZooKeeper - Client environment:user.home=C:\Documents and Settings\Administrator
2013-01-16 15:26:04:INFO org.apache.zookeeper.ZooKeeper - Client environment:user.dir=D:\workspace\stormdemo1
2013-01-16 15:26:04:INFO org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=192.168.0.138 sessionTimeout=2000 watcher=com.zoo.demo.ZooKeeperOperator@ca8327
2013-01-16 15:26:04:INFO org.apache.zookeeper.ClientCnxn - Opening socket connection to server /192.168.0.138:2181
2013-01-16 15:26:13:INFO org.apache.zookeeper.ClientCnxn - Socket connection established to 192.168.0.138/192.168.0.138:2181, initiating session
2013-01-16 15:26:13:INFO org.apache.zookeeper.ClientCnxn - Session establishment complete on server 192.168.0.138/192.168.0.138:2181, sessionid = 0x13c3c5224cc0003, negotiated timeout = 4000
2013-01-16 15:26:13:DEBUG com.zoo.demo.ZooKeeperOperator - 获取设置的信息：ZooKeeper的Java API测试
节点孩子信息:
2013-01-16 15:26:13:DEBUG com.zoo.demo.ZooKeeperOperator - /root中存在节点
2013-01-16 15:26:13:DEBUG com.zoo.demo.ZooKeeperOperator - 节点为：child1
2013-01-16 15:26:13:DEBUG com.zoo.demo.ZooKeeperOperator - 节点为：child3
2013-01-16 15:26:13:DEBUG com.zoo.demo.ZooKeeperOperator - 节点为：child2
2013-01-16 15:26:13:INFO org.apache.zookeeper.ZooKeeper - Session: 0x13c3c5224cc0003 closed
2013-01-16 15:26:13:INFO org.apache.zookeeper.ClientCnxn - EventThread shut down



　　
　　针对出现的中文乱码看采用下面的处理即可：
   String zktest=&quot;ZooKeeper的Java API测试&quot;;
zkoperator.create(&quot;/root/child5&quot;, zktest.getBytes(&quot;utf-8&quot;));
log.debug(&quot;获取设置的信息：&quot;+new String(zkoperator.getData(&quot;/root/child5&quot;),&quot;utf-8&quot;));