目前主流有三种方法去操作zookeeper节点，即zookeeper shell、zookeeper原生API、apache curator API（在原生基础上封装，更加友好易用）。本文主要是基于zookeeper原生API来创建删除查询更改zookeeper节点。
注意事项：
- 原生zookeeper api创建某个节点，必须保证父节点已经存在，否则不能创建（即不能递归创建节点）
- 原生zookeeper api删除某个节点，如果该节点中还有子节点，则该父节点不能直接删除，而必须先删除所有子节点（即不能递归删除节点）
pom中添加如下依赖：
<!-- 如果代码中不使用hadoop 的Configuration，则不需要添加hadoop-common依赖 -->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>2.6.0-cdh5.7.0</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>
</dependency>
java详细代码：
package com.example.zkaccess;
import org.apache.hadoop.conf.Configuration;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
public class TestZk {
// 初始化zookeeper client
public ZooKeeper initZooKeeper(Configuration conf) {
ZooKeeper zooKeeper = null;
try {
zooKeeper = new ZooKeeper(conf.get("ha.zookeeper.quorum"), 3000, new Watcher() {
// conf.get("ha.zookeeper.quorum"): 读取hadoop配置项zookeeper quorum，即常说的connectstring
// 可以配置在core-site.xml中，比如127.0.0.1:2181, 100.11.12.1:2181
// 3000指timeout为3s
@Override
public void process(WatchedEvent event) {
System.out.println(event.toString());
}
});
} catch (IOException e) {
e.printStackTrace();
}
return zooKeeper;
}
// 创建父节点
private void createParentNode(ZooKeeper client, String path) throws KeeperException, InterruptedException {
String newPath = path;
// 删除节点路径首尾的/符号
if(path.startsWith("/")){
newPath = newPath.substring(1);
}
if(path.endsWith("/")){
newPath = newPath.substring(0,newPath.length()-1);
}
System.out.println(">>>>>> newPath is: " + newPath);
String[] pathSplitArr = newPath.split("/");
String subPath = "";
// 逐级创建节点，pathSplitArr.length-1级（第pathSplitArr.length级是seq-node，这里不创建）
for(int i=0;i<pathSplitArr.length-1;i++){
subPath = String.format("%s/%s", subPath, pathSplitArr[i]);
System.out.println(subPath);
if(client.exists(subPath,false)==null) {
// 判断节点是否已经存在，null指不存在，不存在才能创建
client.create(subPath, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
// 参数1:节点路径，参数2:节点数据，参数3:acl，参数4:节点类型
}
}
}
// 创建seq-node
public void createSeqNode(ZooKeeper client, String path) throws KeeperException, InterruptedException {
// 创建String path最后一级的父节点，比如path是"/ns-1/tenant/mysql1/seq-"，那么此处的父节点就是：ns-1、tenant、mysql1
createParentNode(client, path);
// 节点路径必须以“/”开始且不以“/”结束
String newPath = path;
if(!path.startsWith("/")){
newPath = "/".concat(newPath);
}
if(path.endsWith("/")){
newPath = newPath.substring(0, newPath.length()-1);
}
if(client.exists(newPath, false) == null){
// 创建顺序节点，比如"/ns-1/tenant/mysql1/seq-0000000000"
client.create(newPath,"test".getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
}
}
// 读取节点数据
public byte[] getNodeData(ZooKeeper client, String path) throws KeeperException, InterruptedException {
String newPath = path;
byte[] data = null;
if(!path.startsWith("/")){
newPath = "/".concat(newPath);
}
if(path.endsWith("/")){
newPath = newPath.substring(0, newPath.length()-1);
}
System.out.println(">>>>>> getNodeData newPath: " + newPath); Stat stat = client.exists(newPath, false);
System.out.println(stat==null ? ">>>>>> stat null":">>>>>> stat not null");
if(stat != null){
data = client.getData(newPath,false, stat);
}
return data;
}
// 修改节点数据
public boolean setNodeData(ZooKeeper client, String path, byte[] data) {
boolean finished = true;
String newPath = path;
if(!path.startsWith("/")){
newPath = "/".concat(newPath);
}
if(path.endsWith("/")){
newPath = newPath.substring(0, newPath.length()-1);
}
try {
Stat stat = client.exists(newPath, false);
// 节点存在才能够修改节点数据
if( stat != null){
client.setData(newPath, data, stat.getVersion());
// 第3个参数：matched version，在此处为修改前的version，即stat.getVersion()
}
} catch (KeeperException e) {
e.printStackTrace();
finished = false;
} catch (InterruptedException e) {
e.printStackTrace();
finished = false;
}
return finished;
}
// 删除节点
public void deleteNode(ZooKeeper client, String path) throws KeeperException, InterruptedException {
String newPath = path;
if(!path.startsWith("/")){
newPath = "/".concat(newPath);
}
if(path.endsWith("/")){
newPath = newPath.substring(0, newPath.length()-1);
}
Stat stat = client.exists(newPath, false);
// 先删除子节点
if(stat != null){
for(String node : client.getChildren(newPath, false)){
// getChildren获取子节点名集合（不是全路径）
System.out.println(">>>>>> node: " + node);
Stat stat1 = client.exists(String.format("%s/%s", newPath, node),false);
client.delete(String.format("%s/%s", newPath, node), stat1.getVersion());
}
// 再删除父节点
client.delete(newPath, stat.getVersion());
}
}
}
package com.example.zkaccess;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Iterator;
/**
* main类
**/
public class TestMain {
public static void main(String[] args) throws InterruptedException, KeeperException {
// hadoop的配置类Configuration
Configuration conf = new Configuration();
String path = "/ns-1/tenant/mysql1/seq-/";
String path1 = "/ns-1/tenant/mysql1/";
TestZk testZk = new TestZk();
// 初始化zookeeper client对象
ZooKeeper zooKeeper = testZk.initZooKeeper(conf);
// 创建顺序节点
testZk.createSeqNode(zooKeeper, path1);
System.out.println(new String(testZk.getNodeData(zooKeeper, path1), "UTF-8"));
// 修改节点数据
testZk.setNodeData(zooKeeper, path1, "test1".getBytes());
System.out.println(new String(testZk.getNodeData(zooKeeper, path1), "UTF-8"));
// 删除节点
testZk.deleteNode(zooKeeper, path1);
// 关闭zookeeper client
zooKeeper.close();
}
}
网友评论