zookeeper系列(二)zookeeper的使用--javaAPI

时间:2022-11-04 15:49:24

作者:leesf    掌控之中,才会成功;掌控之外,注定失败; 出处:http://www.cnblogs.com/leesf456/ (尊重原创,感谢作者整理的这么好,作者的部分内容添加了我的理解和阐述,供大家一起学习讨论)

一、前言

  上一篇博客我们通过命令行来操作Zookeper的客户端和服务端并进行相应的操作,这篇主要介绍如何通过API(JAVA)来操作Zookeeper。

二、开发环境配置

  首先打开Zookeeper服务端(上一篇博客有具体的方法),方便客户端连接。

  配置开发环境环境可以有两种方式:① 直接下载相关的依赖Jar包,然后在IDE中添加依赖 ② 建立maven项目,使用maven进行依赖管理。

  ① 手动添加依赖至IDE

  步骤一:点击这里下载对应版本的Jar包,包括(jar、javadoc.jar、sources.jar),笔者对应下载的Zookeeper3.4.6版本。

  步骤二:打开IDE(我用的是MyEclipse),新建名为zookeeper_examples_none_maven的java项目。由于需要单独添加依赖,为了方便管理,笔者在项目下新建了jar文件夹,用于存放本项目的jar包(将步骤一下载的3个jar包存放至此文件夹下)。

zookeeper系列(二)zookeeper的使用--javaAPI

  步骤三:在MyEclipse中添加依赖,选中项目右键-->build path -->Libraries-->Add JARs引入jar文件夹下的jar包;(需要加入log4j和slf4j-api俩个jar包)

  zookeeper系列(二)zookeeper的使用--javaAPI

步骤四:新建包、Java类进行测试

  Zookeeper_Constructor_Usage_Simple.java

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
/**
* 使用zookeeper的javaAPI实现客户端对zookeeper服务端的连接;
*/
public class Zookeeper_Constructor_Usage_Simple implements Watcher { private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); ZooKeeper zookeeper;
/**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用connectedSemaphore.await()来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 当客户端连接上了zookeeper服务器,Watcher接口会使用process()方法接收一个连接成功的事件,
* 接下来调用CountDownLatch释放之前的阻塞;
*/
public void process(WatchedEvent event) {
System.out.println("Receive watched event : " + event);
if (KeeperState.SyncConnected == event.getState()) {
System.out.println("连接成功:不再阻塞当前线程");
connectedSemaphore.countDown();
}
} public static void main(String[] args) throws IOException, InterruptedException {
Zookeeper_Constructor_Usage_Simple zookeeperConstructor = new Zookeeper_Constructor_Usage_Simple();
zookeeperConstructor.connect("127.0.0.1:2181");
} }

打印结果为下图所示:

zookeeper系列(二)zookeeper的使用--javaAPI

表示客户端已经成功连接至服务器了。

  可以看到方法一相对而言比较麻烦,需要手动管理不同的依赖jar包,可以采用更成熟的依赖管理方法,即使用maven来管理Jar包。

  ② 使用maven管理依赖

  步骤一:新建maven项目,使用IDE工具直接创建即可;

zookeeper系列(二)zookeeper的使用--javaAPI

直接next即可:

zookeeper系列(二)zookeeper的使用--javaAPI

依然是Next:

zookeeper系列(二)zookeeper的使用--javaAPI

依然是Next:

zookeeper系列(二)zookeeper的使用--javaAPI

步骤二:配置pom.xml文件如下

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.hust.grid.leesf</groupId>
<artifactId>zookeeper_examples</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>jar</packaging>
<name>zookeeper_examples</name>
<url>http://maven.apache.org</url>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.6</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<type>bundle</type>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
</dependencies>
</project>

步骤三:新建java类进行测试

  Zookeeper_Constructor_Usage_Simple.java,代码同上(执行结果不在重复展示)。

先说一下CountDownLatch这个类,这是个很牛逼的门闩,可以对当前执行线程进行管理;我使用该类的源码做一个说明,模拟军队撤退的过程;

package com.hust.grid.leesf.curator;

import java.util.concurrent.CountDownLatch;

 /**
* 模拟军队撤退过程:军官先跑,然后小兵再跑
* CountDownLatch类似于一个门闩,当计数器不为0,当前线程不能执行,使用await方法将线程关在门外;
* 执行一次countDown,计数器就减1,一旦计数器减完后是0,那么await方法立即打开门,让线程执行;
* @author songzl
*
*/
public class Driver {
public static void main(String[] arg) throws InterruptedException {
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(5); for (int i = 0; i < 5; ++i){ // create and start threads
new Thread(new Worker(startSignal, doneSignal,i)).start();
}
//军座先跑
doSomethingElse("军座先撤退...");
//然后计算器减1
startSignal.countDown();//军座撤退完毕,打开门闩,让小兵开始跑
doSomethingElse("军座撤退成功,让小兵撤退...");
//清点人数,确认小兵是否全部撤退,若没有需要列队等待
doneSignal.await(); // wait for all to finish
System.out.println("全军撤退完毕!");
} private static void doSomethingElse(String commond) {
System.out.println(commond);
}
}
class Worker implements Runnable {
private final CountDownLatch startSignal;
private final CountDownLatch doneSignal;
private int count = 0; public Worker(CountDownLatch startSignal, CountDownLatch doneSignal,int count) {
this.startSignal = startSignal;
this.doneSignal = doneSignal;
this.count = count;
}
public void run() {
try {
System.out.println("第"+count+"个小兵试图逃跑,被要求滚回战场...");
startSignal.await();
doWork(count);
doneSignal.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private void doWork(int count) {
System.out.println("军座都跑完了,我们小兵也跑吧,第"+count+"个小兵逃跑:"+Thread.currentThread().getName());
} }

  

三、操作示例

  3.1  创建节点

  创建节点有异步和同步两种方式。无论是异步或者同步,Zookeeper都不支持递归调用,即无法在父节点不存在的情况下创建一个子节点,如在/zk-ephemeral节点不存在的情况下创建/zk-ephemeral/ch1节点;并且如果一个节点已经存在,那么创建同名节点时,会抛出NodeExistsException异常。

  ① 同步方式

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.ACL;
/**
* 使用zookeeper的javaAPI实现客户端对zookeeper服务端的连接;并且给服务端创建znode节点
*/
public class Zookeeper_Constructor_Usage_Simple implements Watcher { private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); ZooKeeper zookeeper;
/**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 当客户端连接上了zookeeper服务器,Watcher接口会使用process()方法接收一个连接成功的事件,
* 接下来调用CountDownLatch释放之前的阻塞;
*/
public void process(WatchedEvent event) {
System.out.println("Receive watched event : " + event);
if (KeeperState.SyncConnected == event.getState()) {
System.out.println("连接成功:不再阻塞当前线程");
connectedSemaphore.countDown();
}
}
/**
* 同步创建节点:方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
Zookeeper_Constructor_Usage_Simple zookeeperConstructor = new Zookeeper_Constructor_Usage_Simple();
zookeeperConstructor.connect("127.0.0.1:2181");
System.out.println("开始给zookeeper服务端创建节点=========");
zookeeperConstructor.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
zookeeperConstructor.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
} }

运行结果如下:

zookeeper系列(二)zookeeper的使用--javaAPI

结果表明已经成功创建了临时节点和临时顺序节点,在创建顺序节点时,系统会在后面自动增加一串数字。

② 异步方式

  使用异步方式于同步方式的区别在于节点的创建过程(包括网络通信和服务端的节点创建过程)是异步的,在同步接口调用过程中,开发者需要关注接口抛出异常的可能,但是在异步接口中,接口本身不会抛出异常,所有异常都会在回调函数中通过Result Code来体现。

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback;
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.data.ACL;
import org.apache.zookeeper.ZooKeeper;
/**
* 使用zookeeper的javaAPI实现客户端对zookeeper服务端的连接,并且使用异步方法创建znode节点
* @author songzl
*/
public class Zookeeper_Create_API_ASync_Usage implements Watcher { private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); ZooKeeper zookeeper;
/**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 当客户端连接上了zookeeper服务器,Watcher接口会使用process()方法接收一个连接成功的事件,
* 接下来调用CountDownLatch释放之前的阻塞;
*/
public void process(WatchedEvent event) {
System.out.println("Receive watched event : " + event);
if (KeeperState.SyncConnected == event.getState()) {
System.out.println("连接成功:不再阻塞当前线程");
connectedSemaphore.countDown();
}
}
/**
* 异步创建节点:方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @param string 一个java的字符串
* @param iStringCallback 回调函数
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode, IStringCallback iStringCallback, String string) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
} public static void main(String[] args) throws Exception {
Zookeeper_Create_API_ASync_Usage zookeeperCreateAPIASyncUsage = new Zookeeper_Create_API_ASync_Usage();
zookeeperCreateAPIASyncUsage.connect("127.0.0.1:2181");
//异步创建临时节点
zookeeperCreateAPIASyncUsage.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL,
new IStringCallback(), "I am context. ");
//异步创建临时顺序节点
zookeeperCreateAPIASyncUsage.create("/zk-test-ephemeral-", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,
new IStringCallback(), "I am context. ");
//如果不阻塞线程,session超时后临时节点会自动删除
Thread.sleep(Integer.MAX_VALUE);
} }
class IStringCallback implements AsyncCallback.StringCallback {
public void processResult(int rc, String path, Object ctx, String name) {
System.out.println("Create path result: [" + rc + ", " + path + ", " + ctx + ", real path name: " + name);
}
}

运行结果如下:

zookeeper系列(二)zookeeper的使用--javaAPI

3.2 删除节点 

  只允许删除叶子节点,即一个节点如果有子节点,那么该节点将无法直接删除,必须先删掉其所有子节点。同样也有同步和异步两种方式。 

  同步和异步方式

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.AsyncCallback;
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.data.ACL;
import org.apache.zookeeper.ZooKeeper; public class Delete_API_Sync_Usage implements Watcher{ private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zookeeper; /**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 当客户端连接上了zookeeper服务器,Watcher接口会使用process()方法接收一个连接成功的事件,
* 接下来调用CountDownLatch释放之前的阻塞;
*/
public void process(WatchedEvent event) {
if (KeeperState.SyncConnected == event.getState()) {
if (EventType.None == event.getType() && null == event.getPath()) {
connectedSemaphore.countDown();
}
}
}
/**
* 方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
/**
* 同步删除znode的节点方法
* @param path znode节点的绝对路径
* @param version znode节点的版本号(-1表示不匹配版本号)
* @throws InterruptedException
* @throws KeeperException
*/
public void syncDelete(String path,int version) throws InterruptedException, KeeperException{
zookeeper.delete(path, version);
}
/**
* 异步删除znode的节点方法
* @param path znode节点的绝对路径
* @param version znode节点的版本号(-1表示不匹配版本号)
* @param cb
* @param ctx
* @throws InterruptedException
* @throws KeeperException
*/
public void AsyncDelete(String path, int version, VoidCallback cb, Object ctx) throws InterruptedException, KeeperException{
zookeeper.delete(path, version, cb, ctx);
} public static void main(String[] args) throws Exception {
String path = "/zk-book";
Delete_API_Sync_Usage delete_API_Sync_Usage = new Delete_API_Sync_Usage();
delete_API_Sync_Usage.connect("127.0.0.1:2181");
//创建path为"/zk-book"的节点
delete_API_Sync_Usage.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("success create znode: " + path);
//创建path为"/zk-book/c1"的节点
delete_API_Sync_Usage.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("success create znode: " + path + "/c1");
//============调用同步删除节点开始=======================================
try {
//直接删除path为"/zk-book"的节点失败,因为包含子节点
delete_API_Sync_Usage.syncDelete(path, -1);
} catch (Exception e) {
System.out.println("fail to delete znode: " + path);
}
//先删除子节点
delete_API_Sync_Usage.syncDelete(path + "/c1", -1);
System.out.println("success delete znode: " + path + "/c1");
//再删除父节点
delete_API_Sync_Usage.syncDelete(path, -1);
System.out.println("success delete znode: " + path);
//============调用同步删除节点结束======================================= //============调用异步步删除节点开始=======================================
delete_API_Sync_Usage.AsyncDelete(path, -1, new IVoidCallback(), null);
delete_API_Sync_Usage.AsyncDelete(path + "/c1", -1, new IVoidCallback(), null);
delete_API_Sync_Usage.AsyncDelete(path, -1, new IVoidCallback(), null); Thread.sleep(Integer.MAX_VALUE);
}
}
/**
* 异步删除时,接口的一个回调函数
* 参数rc表示返回码;
* 参数path和ctx与客户端调用的方法中的参数相等,这两个参数通常用来确定回调中获得的响应是来自于哪个请求的;
* 参数ctx可以是任意对象,只有当path参数不能消灭请求的歧义时才会用到,如果不需要参数ctx,可以设置为null
*/
class IVoidCallback implements AsyncCallback.VoidCallback {
public void processResult(int rc, String path, Object ctx) {
System.out.println(rc + ", " + path + ", " + ctx);
}
}

  备注:ZooKeeper的API提供一个delete()方法来删除一个znode。我们通过输入znode的path和版本号(version number)来删除想要删除的znode。除了使用path来定位我们要删除的znode,还需要一个参数是版本号。只有当我们指定要删除的本版号,与znode当前的版本号一致时,ZooKeeper才允许我们将znode删除掉。这是一种optimistic locking(乐观锁)机制,用来处理znode的读写冲突。我们也可以忽略版本号一致检查,做法就是版本号赋值为-1。

3.3 子节点获取

  读取节点的子节点列表,同样可以使用同步和异步的方式进行操作。

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch; 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.data.ACL;
import org.apache.zookeeper.ZooKeeper;
/**
* 获取zookeeper的子节点:创建一个父节点和三个子节点
*/
public class Zookeeper_GetChildren_API_Sync_Usage implements Watcher { private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zookeeper;
/**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
public void process(WatchedEvent event) {
if (KeeperState.SyncConnected == event.getState()) {
if (EventType.None == event.getType() && null == event.getPath()) {
connectedSemaphore.countDown();
} else if (event.getType() == EventType.NodeChildrenChanged) {
try {
//获取子节点
System.out.println("ReGet Child:" + zookeeper.getChildren(event.getPath(), true));
} catch (Exception e) {
}
}
}
}
/**
* 方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
public static void main(String[] args) throws KeeperException, InterruptedException, IOException {
Zookeeper_GetChildren_API_Sync_Usage zookeeper_GetChildren_API_Sync_Usage = new Zookeeper_GetChildren_API_Sync_Usage();
zookeeper_GetChildren_API_Sync_Usage.connect("127.0.0.1:2181");
String path = "/zk-book-1";
//创建父节点
zookeeper_GetChildren_API_Sync_Usage.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//创建子节点
zookeeper_GetChildren_API_Sync_Usage.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
List<String> childrenList = zookeeper.getChildren(path, true);
System.out.println(childrenList);
zookeeper_GetChildren_API_Sync_Usage.create(path + "/c2", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
Thread.sleep(1000);
zookeeper_GetChildren_API_Sync_Usage.create(path + "/c3", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
Thread.sleep(Integer.MAX_VALUE);
} }

值得注意的是,Watcher通知是一次性的,即一旦触发一次通知后,该Watcher就失效了,因此客户端需要反复注册Watcher,即程序中在process里面又注册了Watcher,否则,将无法获取c3节点的创建而导致子节点变化的事件。

3.4 数据节点获取(不再展示异步的获取方式)

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; 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.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.Stat;
/**
* 数据节点获取
*/
public class GetData_API_Sync_Usage implements Watcher{ private static final int SESSION_TIMEOUT = 5000; private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zookeeper; private static Stat stat = new Stat(); /**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
public void process(WatchedEvent event) {
if (KeeperState.SyncConnected == event.getState()) {
if (EventType.None == event.getType() && null == event.getPath()) {
connectedSemaphore.countDown();
} else if (event.getType() == EventType.NodeDataChanged) {//NodeDeleted表示本节点被变动
try {
System.out.println("the data of znode " + event.getPath() + " is : " + new String(zookeeper.getData(event.getPath(), true, stat)));
System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
} catch (Exception e) {
}
}
}
}
/**
* 方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
public static void main(String[] args) throws Exception, InterruptedException {
String path = "/zk-book";
GetData_API_Sync_Usage getData_API_Sync_Usage = new GetData_API_Sync_Usage();
getData_API_Sync_Usage.connect("127.0.0.1:2181");
//create启动的观察模式
zookeeper.create(path, "123".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println("success create znode: " + path); System.out.println("the data of znode " + path + " is : " + new String(zookeeper.getData(path, true, stat)));
System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
//setData启动的观察模式
zookeeper.setData(path, "123".getBytes(), -1); Thread.sleep(Integer.MAX_VALUE);
}
}

3.5 更新数据

  在更新数据时,setData方法存在一个version参数,其用于指定节点的数据版本,表明本次更新操作是针对指定的数据版本进行的,但是,在getData方法中,并没有提供根据指定数据版本来获取数据的接口,那么,这里为何要指定数据更新版本呢,这里方便理解,可以等效于CAS(compare and swap),对于值V,每次更新之前都会比较其值是否是预期值A,只有符合预期,才会将V原子化地更新到新值B。Zookeeper的setData接口中的version参数可以对应预期值,表明是针对哪个数据版本进行更新,假如一个客户端试图进行更新操作,它会携带上次获取到的version值进行更新,而如果这段时间内,Zookeeper服务器上该节点的数据已经被其他客户端更新,那么其数据版本也会相应更新,而客户端携带的version将无法匹配,无法更新成功,因此可以有效地避免分布式更新的并发问题

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; 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.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.Stat; public class SetData_API_Sync_Usage implements Watcher { private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zookeeper; /**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
public void process(WatchedEvent event) {
if (KeeperState.SyncConnected == event.getState()) {
if (EventType.None == event.getType() && null == event.getPath()) {
connectedSemaphore.countDown();
}
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
SetData_API_Sync_Usage setData_API_Sync_Usage = new SetData_API_Sync_Usage();
setData_API_Sync_Usage.connect("127.0.0.1:2181");
String path = "/zk-book";
setData_API_Sync_Usage.create(path, "songzl".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zookeeper.getData(path, true, null);
Stat stat = zookeeper.setData(path, "wangxn".getBytes(), -1);
System.out.println("czxID: " + stat.getCzxid() + ", mzxID: " + stat.getMzxid() + ", version: " + stat.getVersion());
Stat stat2 = zookeeper.setData(path, "songcy".getBytes(), stat.getVersion());
System.out.println("czxID: " + stat2.getCzxid() + ", mzxID: " + stat2.getMzxid() + ", version: " + stat2.getVersion());
try {
zookeeper.setData(path, "456".getBytes(), stat.getVersion());
} catch (KeeperException e) {
System.out.println("Error: " + e.code() + "," + e.getMessage());
}
Thread.sleep(Integer.MAX_VALUE);
}
}

  运行结果如下 

success create znode: /zk-book
czxID: 2936, mzxID: 2937, version: 1
czxID: 2936, mzxID: 2938, version: 2
Error: BADVERSION,KeeperErrorCode = BadVersion for /zk-book

  结果表明由于携带的数据版本不正确,而无法成功更新节点。其中,setData中的version参数设置-1含义为客户端需要基于数据的最新版本进行更新操作。

3.6 检测节点是否存在

  在调用接口时注册Watcher的话,还可以对节点是否存在进行监听,一旦节点被创建、被删除、数据更新,都会通知客户端。

package com.hust.grid.leesf.examples;

import java.io.IOException;
import java.util.ArrayList;
import java.util.concurrent.CountDownLatch; 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.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.Stat; /**
* 检查节点是否存在
*/
public class Exist_API_Sync_Usage implements Watcher{ private static final int SESSION_TIMEOUT = 5000;
//对执行中的线程进行管理,等待线程完成某些操作后,再对此线程做处理(起到过河拆桥、卸磨杀驴的作用)
private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zookeeper; /**
* 在连接函数中创建了zookeeper的实例;然后建立与服务器的连接;
* 建立连接函数会立即返回,所以我们需要等待连接建立成功后再进行其他的操作;
* 我们使用CountDownLatch来阻塞当前线程,直到zookeeper准备就绪;
*/
public void connect(String host) throws IOException, InterruptedException{
/**
* host 127.0.0.1:2181,服务器端主机名称以及端口号;
* SESSION_TIMEOUT 客户端连接服务器session的超时时间;
* this 表示Watcher接口的一个实例,Watcher实例负责接收Zookeeper数据变化时产生的事件回调;
*/
zookeeper = new ZooKeeper(host, SESSION_TIMEOUT, this);
System.out.println("zk的连接状态:"+zookeeper.getState());
connectedSemaphore.await();
System.out.println("Zookeeper session established");
}
/**
* 方法中调用zookeeper实例的create()方法来创建一个znode;
* @param path znode节点的绝对路径
* @param bytes znode节点的内容(一个二进制数组)
* @param ACL access control list(ACL,访问控制列表,这里使用完全开放模式)
* @param createMode znode的性质,分为EPHEMERAL(临时)、PERSISTENT(持久)、EPHEMERAL_SEQUENTIAL临时顺序和PERSISTENT_SEQUENTIAL持久顺序
* @throws KeeperException
* @throws InterruptedException
*/
public void create(String path,byte[] bytes,ArrayList<ACL> ACL,CreateMode createMode) throws KeeperException, InterruptedException{
String znodePath = zookeeper.create(path, bytes, ACL, createMode);
System.out.println("Success create znode: " + znodePath);
}
/**
* 当客户端连接上了zookeeper服务器,Watcher接口会使用process()方法接收一个连接成功的事件,
* 接下来调用CountDownLatch释放之前的阻塞;若对节点有变动也会根据状态触发该事件;
*/
public void process(WatchedEvent event) {
try {
if (KeeperState.SyncConnected == event.getState()) {//判断状态是否已成功连接
if (EventType.None == event.getType() && null == event.getPath()) {
connectedSemaphore.countDown();
System.out.println("连接服务器成功,关闭阻塞线程");
} else if (EventType.NodeCreated == event.getType()) {//创建节点时触发
System.out.println("success create znode: " + event.getPath());
zookeeper.exists(event.getPath(), true);
} else if (EventType.NodeDeleted == event.getType()) {//删除节点时触发
System.out.println("success delete znode: " + event.getPath());
zookeeper.exists(event.getPath(), true);
} else if (EventType.NodeDataChanged == event.getType()) {//变更当前节点时触发
System.out.println("data changed of znode: " + event.getPath());
zookeeper.exists(event.getPath(), true);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
String path = "/zk-book";
Exist_API_Sync_Usage exist_API_Sync_Usage = new Exist_API_Sync_Usage();
exist_API_Sync_Usage.connect("127.0.0.1:2181");
//若节点不存在返回null,否则返回节点的状态,参数传true是给节点添加观察事件,当节点变动时触发观察事件
Stat stat = zookeeper.exists(path, true);
zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
zookeeper.setData(path, "123".getBytes(), -1); zookeeper.create(path + "/c1", "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
System.out.println("success create znode: " + path + "/c1"); zookeeper.delete(path + "/c1", -1);
zookeeper.delete(path, -1); Thread.sleep(Integer.MAX_VALUE);
} }

运行结果如下:

zookeeper系列(二)zookeeper的使用--javaAPI

结果表明:

  · 无论节点是否存在,都可以通过exists接口注册Watcher。

  · 注册的Watcher,对节点创建、删除、数据更新事件进行监听。

  · 对于指定节点的子节点的各种变化,不会通知客户端(因为指有父节点调用exists了)。

3.7 权限控制

  通过设置Zookeeper服务器上数据节点的ACL控制,就可以对其客户端对该数据节点的访问权限:如果符合ACL控制,则可以进行访问,否则无法访问

  ① 使用无权限信息的Zookeeper会话访问含权限信息的数据节点 

package com.hust.grid.leesf.examples;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper; public class AuthSample_Get {
final static String PATH = "/zk-book-auth_test"; public static void main(String[] args) throws Exception { ZooKeeper zookeeper1 = new ZooKeeper("127.0.0.1:2181", 5000, null);
zookeeper1.addAuthInfo("digest", "foo:true".getBytes());
zookeeper1.create(PATH, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL);
System.out.println("success create znode: " + PATH);
ZooKeeper zookeeper2 = new ZooKeeper("127.0.0.1:2181", 5000, null);
zookeeper2.getData(PATH, false, null);
}
}

运行结果如下:表示权限不够,不能进行操作

success create znode: /zk-book-auth_test
Exception in thread "main" org.apache.zookeeper.KeeperException$NoAuthException: KeeperErrorCode = NoAuth for /zk-book-auth_test
at org.apache.zookeeper.KeeperException.create(KeeperException.java:113)
at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:1155)
at org.apache.zookeeper.ZooKeeper.getData(ZooKeeper.java:1184)
at com.hust.grid.leesf.examples.AuthSample_Get.main(AuthSample_Get.java:17)

② 删除带权限控制的节点

 

package com.hust.grid.leesf.examples;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper; public class AuthSample_Delete {
final static String PATH = "/zk-book-auth_test";
final static String PATH2 = "/zk-book-auth_test/child"; public static void main(String[] args) throws Exception {
ZooKeeper zookeeper1 = new ZooKeeper("127.0.0.1:2181", 5000, null);
zookeeper1.addAuthInfo("digest", "foo:true".getBytes());
zookeeper1.create(PATH, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT); zookeeper1.create(PATH2, "init".getBytes(), Ids.CREATOR_ALL_ACL, CreateMode.EPHEMERAL); try {
ZooKeeper zookeeper2 = new ZooKeeper("127.0.0.1:2181", 5000, null);
zookeeper2.delete(PATH2, -1);
} catch (Exception e) {
System.out.println("fail to delete: " + e.getMessage());
} ZooKeeper zookeeper3 = new ZooKeeper("127.0.0.1:2181", 5000, null);
zookeeper3.addAuthInfo("digest", "foo:true".getBytes());
zookeeper3.delete(PATH2, -1);
System.out.println("success delete znode: " + PATH2); ZooKeeper zookeeper4 = new ZooKeeper("127.00.1:2181", 5000, null);
zookeeper4.delete(PATH, -1);
System.out.println("success delete znode: " + PATH);
}
}

总结:

1、应该使用同步API还是异步API

两种API提供了相同的功能,需要使用哪种API取决于你程序的模式。例如,你设计的程序模式是一个事件驱动模式的程序,那么你最好使用异步API。异步API也可以被用在追求一个比较好的数据吞吐量的场景。想象一下,如果你需要得去大量的znode数据,并且依靠独立的进程来处理他们。如果使用同步API,每次读取操作都会被阻塞住,直到返回结果。不如使用异步API,读取操作可以不必等待返回结果,继续执行。而使用另外的线程来处理返回结果。  

2、观察模式触发器 Watch triggers

读操作,例如:exists、getChildren、getData会在znode上开启观察模式,并且写操作会触发观察模式事件,例如:create、delete和setData。但是ACL(Access Control List)操作不会启动观察模式。观察模式被触发时,会生成一个事件,这个事件的类型取决于触发他的操作:

● exists启动的观察模式,由创建znode,删除znode和更新znode操作来触发。

● getData启动的观察模式,由删除znode和更新znode操作触发。创建znode不会触发,是因为getData操作成功的前提是znode必须已经存在。

● getChildren启动的观察模式,由子节点创建和删除,或者本节点被删除时才会被触发。我们可以通过事件的类型来判断是本节点被删除还是子节点被删除:NodeChildrenChanged表示子节点被删除,而NodeDeleted表示本节点删除。