1. 程式人生 > >Java實現與ZooKeeper的連線

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+"中沒有節點");
			}else{
				log.debug(path+"中存在節點");
				for(String child:list){
					log.debug("節點為:"+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("192.168.0.138");
	            
	            byte[] data = new byte[]{'a','b','c','d'};   
	               
//	            zkoperator.create("/root",null);   
//	            System.out.println(Arrays.toString(zkoperator.getData("/root")));   
//	               
//	            zkoperator.create("/root/child1",data);   
//	            System.out.println(Arrays.toString(zkoperator.getData("/root/child1")));   
//	               
//	            zkoperator.create("/root/child2",data);   
//	            System.out.println(Arrays.toString(zkoperator.getData("/root/child2")));   
	               
	            String zktest="ZooKeeper的Java API測試";
	            zkoperator.create("/root/child3", zktest.getBytes());
	            log.debug("獲取設定的資訊:"+new String(zkoperator.getData("/root/child3")));
	            
	            System.out.println("節點孩子資訊:");   
	            zkoperator.getChild("/root");   
	               
	            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 [email protected]
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="ZooKeeper的Java API測試";
	            zkoperator.create("/root/child5", zktest.getBytes("utf-8"));
	            log.debug("獲取設定的資訊:"+new String(zkoperator.getData("/root/child5"),"utf-8"));