zookeeper集群之java程序同步方式实现对数据节点的增，删，改，查及其权限管理（二）

本文主要介绍通过zookeeper原生的api实现对数据节点的增，删，改，查，及其ACL权限管理。

package com.ilike.zk;
import java.io.IOException;

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.KeeperState;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooKeeper;

/**

* 通过zookeeper客户端同步创建数据节点

*

* @author 桑伟东

*

*

*/

public class CreateNodeSync implements Watcher {

// zookeeper客户端操作类
private static ZooKeeper zooKeeper;

public static void main(String[] args) throws IOException, InterruptedException {
// 服务端的ip地址
String serverIp = "192.168.111.129:2181";
// 会话超时时间
int sessionTimeout = 5000;
// 实例化zookeeper客户端操作类，new CreateSession()为事件监听器对象
zooKeeper = new ZooKeeper(serverIp, sessionTimeout, new CreateNodeSync());
// 读取zookeeper的状态
System.out.println(zooKeeper.getState());// CONNECTING
// 为了让发送的异步请求等到服务器返回的响应后退出，这里让线程睡眠一会儿
Thread.sleep(Integer.MAX_VALUE);
}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
// 输出监听到的事件
System.out.println("接收到事件:" + event);
//// 接收到事件:WatchedEvent state:SyncConnected type:None path:null

// 判断连接是否成功，如果成功，执行我们的业务逻辑
if (event.getState() == KeeperState.SyncConnected) {
doSomething();
}

}

/**
* 在该方法中执行节点的创建 节点的创建模式： PERSISTENT：持久节点
* PERSISTENT_SEQUENTIAL：是一个持久节点，也是一个顺序节点 EPHEMERAL：是一个临时节点
* EPHEMERAL_SEQUENTIAL：是一个临时节点，也是一个顺序节点
*
*/
private void doSomething() {
try {
// 第一个参数为节点路径
// 第二个参数为节点数据
// 第三个参数为ACL权限列表 Ids.OPEN_ACL_UNSAFE表示所有人可读
// 第四个参数为创建模式 CreateMode.PERSISTENT表示这是一个持久节点
String path = zooKeeper.create("/node_2", "中国".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
// 输出服务端返回的节点路径
System.out.println("return path:" + path);// return path:/node_2
} catch (KeeperException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("做一些事情！！！！");
}

}

package com.ilike.zk;

import java.io.IOException;

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.ZooKeeper;

/**

* 通过zookeeper同步删除节点

*

* @author 桑伟东

*

*/

public class DeleteNodeSync implements Watcher {

// zookeeper对象
private static ZooKeeper zooKeeper;

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
// 实例化zookeeper对象
zooKeeper = new ZooKeeper("192.168.111.129:2181", 5000, new DeleteNodeSync());
System.out.println(zooKeeper.getState().toString());

Thread.sleep(Integer.MAX_VALUE);

}

/**
* 删除节点的方法
*
* @param zooKeeper
*/
private void doSomething(ZooKeeper zooKeeper) {
try {
// 第一个参数为需要删除的节点的完整路径
// 第二个参数为版本，-1表示不校验版本信息
zooKeeper.delete("/node_5", -1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub

if (event.getState() == KeeperState.SyncConnected) {
if (event.getType() == EventType.None && null == event.getPath()) {
doSomething(zooKeeper);
}

}
}

}

package com.ilike.zk;
import java.io.IOException;

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.ZooKeeper;

import org.apache.zookeeper.data.Stat;

/**

* 通过zookeeper同步修改节点数据

*

* @author 桑伟东

*

*/

public class UpdateNodeSync implements Watcher {

// zookeeper对象
private static ZooKeeper zooKeeper;

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
// 实例化zookeeper对象
zooKeeper = new ZooKeeper("192.168.111.129:2181", 5000, new UpdateNodeSync());
System.out.println(zooKeeper.getState().toString());

Thread.sleep(Integer.MAX_VALUE);

}

/**
* 修改节点数据的方法
*
* @param zooKeeper
*/
private void doSomething(ZooKeeper zooKeeper) {
try {
Stat stat = zooKeeper.setData("/node_6", "123".getBytes(), -1);
System.out.println("stat:" + stat);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub

if (event.getState() == KeeperState.SyncConnected) {
if (event.getType() == EventType.None && null == event.getPath()) {
doSomething(zooKeeper);
}

}
}

}

package com.ilike.zk;
import java.io.IOException;

import java.util.List;

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.ZooKeeper;

/**

* 通过zookeeper客户端同步获取子节点列表

*

* @author 桑伟东

*

*/

public class GetChildrenSync implements Watcher {

// zookeeper的客户端对象
private static ZooKeeper zooKeeper;

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
// 初始化zookeeper的客户端对象
zooKeeper = new ZooKeeper("192.168.111.129:2181", 5000, new GetChildrenSync());
System.out.println(zooKeeper.getState().toString());

Thread.sleep(Integer.MAX_VALUE);

}

/**
* 获取子节点的方法
*
* @param zooKeeper
*/
private void doSomething(ZooKeeper zooKeeper) {

try {
// 第一个参数为要获取的节点
// 第二个参数为在获取子节点的过程中，要不要关注子节点的变化，需要为true，不需要为false
// 返回子节点的列表
List<String> children = zooKeeper.getChildren("/", true);
// 输出子节点列表
System.out.println(children);// [node_3, node_2, node_1, zookeeper]

} catch (Exception e) {
// TODO: handle exception
}

}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
// 如果连接成功，查询
if (event.getState() == KeeperState.SyncConnected) {
// 刚连接上服务器,事件的类型为空，与事件关联的路径也为空
// 加上这个判断可以保证doSomething()方法只执行一次
if (event.getType() == EventType.None && null == event.getPath()) {
doSomething(zooKeeper);
} else {
// 如果在获取子节点列表的过程中，我们关注子节点的变化，那么在这里我们会收到子节点变化的信息

// 判断该信息是否是子节点变化，如果是，获取变化的节点的子节点
if (event.getType() == EventType.NodeChildrenChanged) {
try {
System.out.println(zooKeeper.getChildren(event.getPath(), true));
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}

}

package com.ilike.zk;
import java.io.IOException;

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.ZooKeeper;

import org.apache.zookeeper.data.Stat;

/**

* 通过zookeeper对象同步获取节点数据

*

* @author 桑伟东

*

*/

public class GetDataSync implements Watcher {

// zookeeper对象
private static ZooKeeper zooKeeper;
private static Stat stat = new Stat();

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
// 实例化zookeeper对象
zooKeeper = new ZooKeeper("192.168.111.129:2181", 5000, new GetDataSync());
System.o
c283
ut.println(zooKeeper.getState().toString());
Thread.sleep(Integer.MAX_VALUE);

}

/**
* 获取节点数据的方法
*
* @param zookeeper
*/
private void doSomething(ZooKeeper zookeeper) {

// zookeeper.addAuthInfo("digest", "jike:123456".getBytes());
try {
// 第一个参数为你需要获取数据的节点
// 第二个参数为是否关注数据变化
// 第三个参数为zookeeper的状态实例，zookeeper会将返回的数据信息封装在这个事例中
System.out.println(new String(zooKeeper.getData("/node_1", true, stat)));
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}

}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub

if (event.getState() == KeeperState.SyncConnected) {
if (event.getType() == EventType.None && null == event.getPath()) {
doSomething(zooKeeper);
} else {
if (event.getType() == EventType.NodeDataChanged) {
try {
System.out.println(new String(zooKeeper.getData(event.getPath(), true, stat)));
System.out.println("stat:" + stat);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}

}
}

}
}

}

package com.ilike.zk;
import java.io.IOException;

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.ZooKeeper;

import org.apache.zookeeper.data.Stat;

/**

* 通过zookeeper同步方式判断节点是否存在

*

* @author 桑伟东

*

*/

public class NodeExistsSync implements Watcher {

private static ZooKeeper zooKeeper;

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

zooKeeper = new ZooKeeper("192.168.111.129:2181", 5000, new NodeExistsSync());
System.out.println(zooKeeper.getState().toString());
Thread.sleep(Integer.MAX_VALUE);
}

/**
* 判断节点是否存在的方法
*
* @param zooKeeper
*/
private void doSomething(ZooKeeper zooKeeper) {

try {
// 第一个参数为要判断的节点的路径
// 第二个参数为是否关注节点变化
Stat stat = zooKeeper.exists("/node_1", true);
System.out.println(stat);

} catch (Exception e) {

}

}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub

if (event.getState() == KeeperState.SyncConnected) {
if (event.getType() == EventType.None && null == event.getPath()) {
doSomething(zooKeeper);
} else {
try {
if (event.getType() == EventType.NodeCreated) {
System.out.println(event.getPath() + " created");
System.out.println(zooKeeper.exists(event.getPath(), true));
} else if (event.getType() == EventType.NodeDataChanged) {
System.out.println(event.getPath() + " updated");
System.out.println(zooKeeper.exists(event.getPath(), true));
} else if (event.getType() == EventType.NodeDeleted) {
System.out.println(event.getPath() + " deleted");
System.out.println(zooKeeper.exists(event.getPath(), true));
}

} catch (Exception e) {
// TODO: handle exception
}

}

}
}

}

package com.ilike.zk;
import java.io.IOException;

import java.security.NoSuchAlgorithmException;

import java.util.ArrayList;

import org.apache.zookeeper.CreateMode;

import org.apache.zookeeper.KeeperException;

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooDefs.Ids;

import org.apache.zookeeper.ZooDefs.Perms;

import org.apache.zookeeper.ZooKeeper;

import org.apache.zookeeper.Watcher.Event.EventType;

import org.apache.zookeeper.Watcher.Event.KeeperState;

import org.apache.zookeeper.data.ACL;

import org.apache.zookeeper.data.Id;

import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;

/**

* zookeeper的权限管理

*

* @author 桑伟东

*

*

* 权限模式(scheme): ip, digest 授权对象(ID)

* ip权限模式: 具体的ip地址 ַ

* digestȨ权限模式:username:Base64(SHA-1(username:password))

* 权限(permission): create(C),DELETE(D),READ(R), WRITE(W), ADMIN(A)

* 注：单个权限，完全权限，符合权限

*

* 权限组合: scheme + ID + permission

*

*

*

*/

public class CreateNodeSyncAuth implements Watcher {

// zookeeper对象
private static ZooKeeper zookeeper;
private static boolean somethingDone = false;

public static void main(String[] args) throws IOException, InterruptedException {
// 实例化zookeeper对象
zookeeper = new ZooKeeper("192.168.111.129:2181", 5000, new CreateNodeSyncAuth());
System.out.println(zookeeper.getState());

Thread.sleep(Integer.MAX_VALUE);
}

private void doSomething() {
try {
// 第一个参数为具体的权限
// 第二个参数为权限模式对象（参数为ip模式及具体的IP地址）
ACL aclIp = new ACL(Perms.READ, new Id("ip", "192.168.111.129"));
ACL aclDigest = new ACL(Perms.READ | Perms.WRITE,
new Id("digest", DigestAuthenticationProvider.generateDigest("swd:8859-1")));
ArrayList<ACL> acls = new ArrayList<ACL>();
acls.add(aclDigest);
acls.add(aclIp);

/*
* 第三个参数可以自己设置，也可以采用zookeeper预设的方式，主要有以下几种
* Ids.OPEN_ACL_UNSAFE:表示所有用户都可以做任何操作
* Ids.CREATOR_ALL_ACL:表示系统将通过addAuthInfo来为新创建的节点添加授权信息
*/
zookeeper.addAuthInfo("digest", "swd:8859-1".getBytes());
// String path = zookeeper.create("/node_5", "8899".getBytes(),
// acls, CreateMode.PERSISTENT);
String path = zookeeper.create("/node_5", "8899".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
System.out.println("return path:" + path);

somethingDone = true;

} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (NoSuchAlgorithmException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
System.out.println("接收到的事件" + event);
if (event.getState() == KeeperState.SyncConnected) {
if (!somethingDone && event.getType() == EventType.None && null == event.getPath()) {
doSomething();
}
}
}

}