Zookeeper学习之源生API的使用(java与shell操作zookeeper)。

如果不会搭建zookeeper环境：请看此文章：Zookeeper学习之集群环境搭建

1.操作zookeeper(shell)

   启动zookeeper客户端：zkCli.sh;

   根据提示命令进行操作：

    查找：

根据提示命令进行操作：
查找：
ls/ 	ls/zookeeper
创建并赋值：
create /adai hello
获取：
get adai
设置：
set /adai world
可以看到zookeeper集群的数据一致性
rmr /path	递归删除节点
delete /path/child  删除指定某个节点

创建节点有两种类型：短暂（ephemeral）  持久（persistent）

2.zookeeper-3.3.4jar简单介绍

zookeeper的api（zookeeper-3.3.4.jar为官方提供的javaAPI,zkClient-0.1.jar为源生api的基础之上进行扩展的开源java客户端）
Zookeeper（Argumetns）方法（一共有4个构造方法,根据参数不同）
参数说明如下：
connectString:链接服务列表，已“,”分割
sessionTimeout:心跳检测时间周期（毫秒）。
wather:事件处理通知器。
canBeReadOnly:标识当前回话是否支持只读。
sessionId和sessionPassword:提供链接zookeeper的sessionId和密码，通过这两个确定唯一一个客户端，目的是可以提供重复会话。

注意：zookeeper客户端和服务器端会话的建立是一个异步的过程，也就是说在程序中，我们程序方法在处理完客户端初始化后立即返回

（也就是说程序往下执行代码，这样，大多数情况下我们并没有真正构建好一个可用会话，在会话的生命周期处于Connecting时才算是真正建立完毕，

所以我们需要使用多线程中学习的小工具CountdownLaunch）

3.java操作zookeeper之api方法详解。

创建节点（znode）方法create
提供了两套创建节点的方法，同步和异步创建节点方式。
同步方式：
参数1：节点路径（名称）：/nodeName(不允许递归创建节点，也就是说节点不存在的情况下，不允许创建子节点)
参数2：节点内容：要求类型是字节数组（也就是说，不支持序列化方式，如果需要实现序列化，可以使用java相关序列化框架，如Hessian,Kryo框架）
参数3：节点权限，使用ids.OPEN_ACL_UNSAPE开放即可（这个参数一般在权限没有太高要求的场景使用，没必要关注）
参数4：节点类型，创建节点的类型：CreateMode.*,提供四种节点类型
PERSISTENT(持久及诶单)
PERSISTENT_SEQUENTIAL（持久顺序节点）
EPHEMERAL（临时节点）
EPHEMERAL_SEQUENTIAL（临时顺序节点）
异步方式：（在同步参数基础上增加两个参数）
参数5：注册一个异步回调函数，要实现AsynCallBack.StringCallBack接口，重写
ProcessResult(int rc , String path , Object ctx , String name)方法，当节点创建完毕后执行此方法。
rc:为服务端响应码0表示调用成功，-4表示端口连接，-110表示执行节点存在，-112表示会话已过期。
path:接口调用时传入API的数据节点的路径参数
ctx：为调用接口传入API的cxt值
name：实际在服务器端创建节点的名称
参数6：传递给回调函数的参数，一般为上下文（Context）信息

删除节点：delete方法（api提供两个接口：同步和异步删除方式）
同步方式：
参数1：节点名称 /deletePath
参数2：版本号，即本次删除操作时针对该数据的某个版本进行的操作。（-1，代表忽略该版本号）
异步方式：（和delete方法一致）
参数3：一个异步回调函数
参数4：用于传递上下文信息的对象
注意：在zookeeper中，只允许删除叶子节点信息，也就是说如果当前节点不是叶子节点则无法删除，或者必须先删除旗下所有子节点。

getChildren读取数据方法：包括子节点列表的获取和子节点数据的获取
参数1：path:获取指定节点下的数据（获取字节点列表）
参数2：watcher：注册的watcher，一旦在本次子节点获取后，子节点列表发生变更的话，那么就会向客户端

发送通知，该参数允许为null.
参数3：watch:表明是否需要注册一个watcher；如果为true，则会使用到zookeeper客户端上文中提到的那个

默认watcher,如果为false,则表明不需要注册Watcher.
参数4：cb：回调函数。
参数5：ctx:上下文信息对象。
参数6：stat:指定数据节点的节点状态信息。

注意：当我们获取指定节点的子节点列表后，还需要订阅这个子节点列表的变换通知，这时候就可以通过注册一个watcher来实现，
当子节点被添加或删除时，服务器端就是会触发一个NodeChildrenChanged类型的事件通知，需要注意的是服务器端发送给客户端
的事件通知中，是不包含最新的节点列表的，客户端必须主动从新进行获取。通常在客户端接收到这个事件后，就可以再次主动获
取最新的子节点列表了。也就是说，zookeeper服务器在向客户端发送NodeChildrenChange事件通知的时候，仅仅只是发了一个通知，
不会把节点变化情况发送个客户端，需要客户端自己重新获取，另外Watcher通知是一次性的，即触发后失效，因此客户端需要反复注册watcherc才行。

getDate方法：获取指定节点的数据内容。
参数1：patch:路径
参数2：watcher:注册的watcher对象。一旦之后及诶单内容更新，则会向客户端发送通知，该参数允许为null.
参数3：status:指定及诶单的状态信息。
参数4：watch是否使用watcher，如果为true,则使用默认上文中的watcher，false则不使用watcher
参数5;cb:回调函数
参数6：ctx：用于传递参数的下文信息对象。setData方法：修改指定节点的数据内容。

参数1：path:路径
参数2：data:数据内容。
参数3：版本号（-1覆盖之前所有的版本）
参数4：cb：回调函数。
参数5: ctx:用于传递的下文信息对象。

注意：该方法和getChildren方法基本相同。主要是注册的watcher有所不同，客户端在获取一个阶段数据内容时，是可以进行watcher注册的，一旦节点发生变更，

则服务器会发送给客户端一个NodeDateChanged的事件通知。

exists方法：节点检测是否存在。
参数1：path：路径
参数2：watcher:注册的watcher对象，一旦之后节点内容有变更，则会像客户端发送通知，该参数允许为null。
参数3：watch:是否使用watcher，如果为true,则使用默认上文中的watcher,false,则不使用

4.java代码操作zookeeper。需要用的jar有：log4j-1.2.15.jar 、slf4j-api-1.6.1.jar、slf4j-log4j12-1.6.1.jar、zookeeper-3.4.5.jar
自定义zookeeper异常类

package com.adai.exception;
/**
* 自定义操作zookeeper异常类
* @author adai
*/
public class ZookeeperRunTimeException extends RuntimeException{
private static final long serialVersionUID = 1L;
private static String msgErr = "客户端操作zookeeper出现异常";

public ZookeeperRunTimeException(){
super(msgErr);
}

public ZookeeperRunTimeException(String mesErr,Object ...ars){
super(mesErr);
for(int i=0;i<ars.length;i++){
System.out.println("ars["+(i+1)+"]:"+ars[i].toString());
}
}
}

操作zookeeper源生api的java类：

package com.adai.zookeeper.one.base;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import com.adai.exception.ZookeeperRunTimeException;
/**
* zookeeperApi学习
* @author adai
*/
public class ZookeeperBase {
//zookeeper地址
private  String CONNECT_ADDR = "192.168.110.135:2181,192.168.110.136:2181,192.168.110.137:2181";
//session超时时间
private  int SESSION_OUTTIME = 2000 ; //ms
//信号量，阻塞程序执行，用于等待zookeeper链接成功，发送成功信号
private  CountDownLatch countDown = new CountDownLatch(1);
private ZooKeeper zoo = null;
public ZookeeperBase(){
try {
zoo = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher() {
@Override
public void process(WatchedEvent event) {
//获取事件状态
KeeperState state = event.getState();
EventType eventType = event.getType();
if(KeeperState.SyncConnected == state){
if(EventType.None == eventType ){ // 刚链接时，没有任何节点
//如果连接成功，则发送信号量，让后续阻塞程序向下执行
System.out.println("客户端连接服务器成功。。。。");
countDown.countDown();
}
}
}
});
//阻塞程序，等待客户端连接成功
countDown.await();
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}

/**
* 创建根节点信息
* @param path 根节点路径
* @param data 节点数据
* @param dataMode 数据存储的类型
* @throws KeeperException
* @throws InterruptedException
*/
public  void createRootNode(String path , String data , CreateMode ...dataMode)
throws KeeperException, InterruptedException{
Stat stat = zoo.exists(path, false);
if(stat != null) return;
//创建父节点节点
String msg = zoo.create(
path, // 节点路径，不能递归创建
data.getBytes(), // 只能是字节数组类型
Ids.OPEN_ACL_UNSAFE, // 节点权限类型
dataMode.length == 0 ? CreateMode.PERSISTENT : dataMode[0]// 节点类型，持久化和临时节点+有序节点
);
System.out.println(path+"节点创建成功。。。。");
System.out.println(msg);
}

/**
* 递归创建子节点
* @param path 子节点全路径
* @param data 子节点内容信息
* @throws KeeperException
* @throws InterruptedException
*/
public void createChilrenNode(String path , String data)
throws KeeperException, InterruptedException{
String[] paths = path.split("/");
String rootPath = paths[0]+"/"+paths[1]; //根节点路径
Stat stat = zoo.exists(rootPath, false);
if(stat == null){ //如果根节点路径不存在，则创建
createRootNode(rootPath,"");
}
for(int i=2;i<paths.length;i++){
String chilrenPath = getChildrenPath(paths , i);
createRootNode(chilrenPath,data);
}
}

/**
* 递归创建节点
* @param path 节点全路径
* @param nodeData 每个节点对应的数据，当节点的数据小于节点路径，缺少几个从根节点开始补空
* @throws KeeperException
* @throws InterruptedException
*/
public void createChilrenNode(String path , String[] nodeData)
throws KeeperException, InterruptedException{
String[] paths = path.split("/");
String[] nodeDatas = minusNodeDate(paths , nodeData);
String rootPath = paths[0]+"/"+paths[1]; //根节点路径
String rootData = nodeDatas[1];
Stat stat = zoo.exists(rootPath, false);
if(stat == null){ //如果根节点路径不存在，则创建
createRootNode(rootPath,rootData);
}
for(int i=2;i<paths.length;i++){
String chilrenPath = getChildrenPath(paths , i);
createRootNode(chilrenPath,nodeDatas[i]);
}
}

/**
* 取得节点对应的数据，如果节点数据小于节点路径，之前的节点数据为空处理
* @param paths 节点的全路径
* @param nodesDate 每个节点对应的节点数据
* @return 所有节点数据
*/
public String[] minusNodeDate(String[] paths , String[] nodesDate){
if(paths.length  == nodesDate.length )
return nodesDate;
if(paths.length < nodesDate.length )
throw new ZookeeperRunTimeException("子节点数据多于节点路径");
String[] tempNode = new String[paths.length];
int minus = paths.length - nodesDate.length;
for(int i=0;i<minus;i++){
tempNode[i] = "";
}
for(int i=0 ; i<nodesDate.length;i++){
tempNode[minus++] = nodesDate[i];
}
return tempNode;
}

/**
* 获取到子节点的全路径
* @param paths
* @param index
* @return
*/
public String getChildrenPath(String[] paths , int index ){
StringBuffer sb = new StringBuffer();
for(int i=1 ; i<=index;i++){
sb.append("/").append(paths[i]);
}
return sb.toString();
}

/**
* 获取节点的数据
* @param nodePath
* @throws InterruptedException
* @throws KeeperException
* @return
*/
public byte[] getNodeData(String nodePath)
throws KeeperException, InterruptedException{
Stat sata = zoo.exists(nodePath, false);
if(sata == null ){
throw new ZookeeperRunTimeException("该节点不存在", nodePath);
}
byte[] data = zoo.getData(nodePath, false, null);
return data;
}

/**
* 得到所有节点数据
* @param nodePath 节点路径
* @return 所有节点的内容
* @throws KeeperException
* @throws InterruptedException
*/
public Object[] getAllNodeData (String nodePath)
throws KeeperException, InterruptedException{
String[] nodePaths = nodePath.split("/");
Object[] nodeDates = new Object[nodePaths.length-1];
for(int i=1;i<nodePaths.length;i++){
String nodeTempPath = getChildrenPath(nodePaths, i);
nodeDates[i-1] = getNodeData(nodeTempPath);
}
return nodeDates;
}

/**
* 判断节点是否存在
* @param nodePath  节点路径
* @return true为存在该子节点
* @throws KeeperException
* @throws InterruptedException
*/
public boolean isExistNode(String nodePath)
throws KeeperException, InterruptedException{
Stat stat = zoo.exists(nodePath, false);
if(stat == null){
return false;
}
return true;
}

/**
* 删除某一个子节点（只能删除子节点）
* @param nodePath 子节
c4a0
点路径
* @throws KeeperException
* @throws InterruptedException
*/
public void deleteNode(String nodePath)
throws KeeperException, InterruptedException{
if(isExistNode(nodePath)){
List<String> children = zoo.getChildren(nodePath, false);
if(children != null && !children.isEmpty() ){
System.out.println("该节点不是子节点，无法删除");
return;
}
String data = new String(getNodeData(nodePath));
zoo.delete(nodePath, -1); // -1删除所有版本
System.out.println("删除成功:"+nodePath+":"+data);
}else{
System.out.println("没有该子节点:"+nodePath);
}
}

/**
* 删除根节点
* @param nodePath 根节点
* @throws KeeperException
* @throws InterruptedException
*/
public void deleteAllNode(String nodePath)
throws KeeperException, InterruptedException{
List<String> nodes = new ArrayList<String>();
getChildrenNode(nodePath,nodes);
nodes.add(0,nodePath); // 将根节点放入容器中
Collections.reverse(nodes); // 倒置
for(String node : nodes){ //递归删除所有子节点
deleteNode(node);
}
}

/**
* 得到根节点下的所有子节点全路径
* @param nodePath 根节点路径
* @param list 存放数据的集合（如果根节点/a 下有/b、/c这两个子节点，/b下有/b1、b2两个子节点，/c有 /c1子节点）
*        </br>的值将会有 /a/c/c1,/a/c,/a/b/b2,/a/b/b1,/a/c,/a/b
* @throws KeeperException
* @throws InterruptedException
*/
public void getChildrenNode(String nodePath ,List<String> list)
throws KeeperException, InterruptedException{
List<String> nodes = zoo.getChildren(nodePath, false);
for(int i = 0 ; i<nodes.size();i++){
String node = nodes.get(i);
list.add(nodePath+"/"+node);
getChildrenNode(nodePath+"/"+node,list);
}
return;
}

/**
* 设置子节点的值
* @param nodePath 子节点的路径
* @param data 子节点需要修改的值
* @throws KeeperException
* @throws InterruptedException
*/
public void setNodeData(String nodePath , String data)
throws KeeperException, InterruptedException{
zoo.setData(nodePath, data.getBytes(), -1);
}

/**
* 修改所有节点的数据内容
* @param nodePath 节点全路径
* @param data 节点对应的数据内容（数组的最后一个值对应，nodePath的最后一个节点）
* @throws KeeperException
* @throws InterruptedException
*/
public void setAllNodeData(String nodePath , String[] data )
throws KeeperException, InterruptedException{
String[] nodePaths = nodePath.split("/");
StringBuffer sb = new StringBuffer();
if(nodePaths.length == data.length ){
for(int i=1;i<nodePath.length() ;i++){
sb.append("/").append(nodePaths[i]);
setNodeData(sb.toString(),data[i]);
}
}
List<String> list = getNodePath(nodePath);
Collections.reverse(list);
for(int i=data.length-1;i>=0;i--){
setNodeData(list.get(data.length -1 - i ),data[i]);
}
}

/**
* 得到所有节点的全路径路径
* @param nodePath  节点路径
* @return 如果nodePath值为：/a/b/c  将返回：/a /a/b /a/b/c
*/
public List<String> getNodePath(String nodePath){
List<String> list = new ArrayList<String>();
StringBuffer sb = new StringBuffer();
String temp[] = nodePath.split("/");
for(int i=1;i<temp.length;i++){
sb.append("/").append(temp[i]);
list.add(sb.toString());
}
return list;
}

/**
*  异步删除子字节
* @param nodePath  节点路径
*/
public void deleteCallBack(String nodePath){
zoo.delete(nodePath, -1, new VoidCallback() {

@Override
public void processResult(int rc, String path, Object ctx) {
/**rc:为服务端响应码0表示调用成功，-4表示端口连接，-110表示执行节点存在，112表示会话已过期。
* path:接口调用时传入API的数据节点的路径参数
* ctx：为调用接口传入API的cxt值name：实际在服务器端创建节点的名称
**/
System.out.println(rc);
System.out.println(path);
System.out.println(ctx);
}
}, "aa");
}

/**
* 释放连接
* @throws InterruptedException
*/
public void close() throws InterruptedException{
zoo.close();
}

public static void main(String[] args) throws Exception {

ZookeeperBase zooBase = new ZookeeperBase();
//		zooBase.createRootNode("testRoot","hello world");

//		zooBase.createChilrenNode("/hello/adai/b","hello adai b");
//
//		zooBase.createChilrenNode("/hello/adai/c","hello adai c");
//		zooBase.createChilrenNode("/a/b/c",new String[] {"a","b","c"});

//		System.out.println(new String ( zooBase.getNodeData("/a/b/c") ) );
//
//		zooBase.createChilrenNode("/a/b1/c1",new String[] {"c1"});
//		System.out.println( zooBase.isExistNode("/a/b/c1") );
//		zooBase.getAllNodeData("/a/b/c1");

//		zooBase.deleteNode("/a");
//		List<String> list = new ArrayList<String>();
//		zooBase.getChildrenNode("/a",list);
//		zooBase.setNodeData("/a", "aac");
//		zooBase.deleteAllNode("/a");
//		zooBase.setAllNodeData("/a/b/c", new String[]{"e","f"});

zooBase.deleteCallBack("/hello");
zooBase.close();
}

}

遗留问题：在使用deleteCallBack方法进行节点删除时，无法将节点删除，原因不清楚。如有知道，麻烦告知一下。

总结：本文主要是练习zookeeper-3.4.5.jar源生api的使用，主要练习的方法有，create,delete,setData,getData等方法。在使用java代码对zookeeper-3.4.5进行操作时，由于

源生的api不支持删除根节点，所有自己新添了一些方法：比如递归的创建子节点、删除根节点等方法，新添这些方法的目的主要是为了更好的理解zookeeper源生api的使用。

如有疑问，或者此文章的代码有问题，希望您告知本人一下。qq:1721995310.

遗留问题解决方法：deleteCallBack方法不能删除根节点，如果删除根节点rc的值为：-111，如果删除节点不存在返回-101,由于该方法为异步删除，可能在删除的过程中主线程

已经执行完成。解决方法为：在主线程调用deleteCallBack方法后，将主线程休眠5秒即可。