zookeeper JavaAPI入门操作

1 安装

# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between
# sending a request and getting an acknowledgement
syncLimit=5
# the directory where the snapshot is stored.
dataDir=/tmp/zookeeper
# the port at which the clients will connect
clientPort=2181

2 启动Server

zkServer.cmd start

3 启动client

zkCli.cmd -server localhost:2181

4 使用Java API 与zookeeper 服务端交互

<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>zookeeper</artifactId>
<version>3.3.0</version>
<exclusions>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>

public class demo {
//会话超时时间，设置为与系统默认时间一致
private static final int SESSION_TIMEOUT = 30000;

//创建ZooKeeper实例
ZooKeeper zk;

//创建Watcher实例
Watcher wh = new Watcher() {
public void process(org.apache.zookeeper.WatchedEvent event) {
System.out.println(event.toString());
}
};

//初始化ZooKeeper实例
private void createZKInstance() throws IOException {
zk = new ZooKeeper("localhost:2181", demo.SESSION_TIMEOUT, this.wh);

}

private void ZKOperations() throws IOException, InterruptedException, KeeperException {
System.out.println("\n1.创建ZooKeeper节点(znode：zoo2,数据：myData2，权限：OPEN_ACL_UNSAFE，节点类型：Persistent");
zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

System.out.println("\n2.查看是否创建成功：");
System.out.println(new String(zk.getData("/zoo2", false, null)));

System.out.println("\n3.修改节点数据");
zk.setData("/zoo2", "shenlan211314".getBytes(), -1);

System.out.println("\n4.查看是否修改成功：");
System.out.println(new String(zk.getData("/zoo2", false, null)));

System.out.println("\n5.删除节点");
zk.delete("/zoo2", -1);

System.out.println("\n6.查看节点是否被删除：");
System.out.println("节点状态：[" + zk.exists("/zoo2", false) + "]");
}

private void ZKClose() throws InterruptedException {
zk.close();
}

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
demo dm = new demo();
dm.createZKInstance();
dm.ZKOperations();
dm.ZKClose();
}
}

5 应用程序实例：创建组，加入组，列出组成员，删除组

创建zookeeper实例

public class ZooKeeperInstance {
//会话超时时间，设置为与系统默认时间一致
public static final int SESSION_TIMEOUT=30000;

//创建ZooKeeper实例
ZooKeeper zk;

//创建Watcher实例
Watcher wh=new Watcher(){
public void process(WatchedEvent event){
System.out.println(event.toString());
}
};

//初始化Zookeeper实例
public void createZKInstance() throws IOException{
zk=new ZooKeeper("10.100.90.183:2181",ZooKeeperInstance.SESSION_TIMEOUT,this.wh);
}

//关闭ZK实例
public void ZKclose() throws InterruptedException{
zk.close();
}
}

创建组

public class CreateGroup extends ZooKeeperInstance {

//创建组
//参数：groupPath
public void createPNode(String groupPath) throws KeeperException, InterruptedException{
//创建组
String cGroupPath=zk.create(groupPath, "group".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//输出组路径
System.out.println("创建的组路径为："+cGroupPath);
}

public static void main(String[] args) throws IOException, KeeperException, InterruptedException{
CreateGroup cg=new CreateGroup();
cg.createZKInstance();
cg.createPNode("/ZKGroup");
cg.ZKclose();
}

}

加入组

public class JoinGroup extends ZooKeeperInstance {
//加入组操作
public int Join(String groupPath,int k) throws KeeperException, InterruptedException{
String child=k+"";
child="child_"+child;

//创建的路径
String path=groupPath+"/"+child;
//检查组是否存在
if(zk.exists(groupPath,true) != null){
//如果存在，加入组
zk.create(path,child.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
return 1;
}
else{
System.out.println("组不存在！");
return 0;
}
}

//加入组操作
public void MultiJoin() throws KeeperException, InterruptedException{
for(int i=0;i<10;i++){
int k=Join("/ZKGroup",i);
//如果组不存在则退出
if(0==k)
System.exit(1);
}
}

public static void main(String[] args) throws IOException, KeeperException, InterruptedException{
JoinGroup jg=new JoinGroup();
jg.createZKInstance();
jg.MultiJoin();
jg.ZKclose();
}
}

列出成员

public class ListMembers extends ZooKeeperInstance {
public void list(String groupPath) throws KeeperException, InterruptedException{
//获取所有子节点
List<String> children=zk.getChildren(groupPath, false);
if(children.isEmpty()){
System.out.println("组"+groupPath+"中没有组成员存在！");
System.exit(1);
}
for(String child:children)
System.out.println(child);
}

public static void main(String[] args) throws IOException, KeeperException, InterruptedException{
ListMembers lm=new ListMembers();
lm.createZKInstance();
lm.list("/ZKGroup");
}
}

删除组

public class CreateGroup extends ZooKeeperInstance {

//创建组
//参数：groupPath
public void createPNode(String groupPath) throws KeeperException, InterruptedException{
//创建组
String cGroupPath=zk.create(groupPath, "group".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//输出组路径
System.out.println("创建的组路径为："+cGroupPath);
}

public static void main(String[] args) throws IOException, KeeperException, InterruptedException{
CreateGroup cg=new CreateGroup();
cg.createZKInstance();
cg.createPNode("/ZKGroup");
cg.ZKclose();
}

}