image.png
Zookeeper系列文章
1.Zookeeper简介
2.Zookeeper集群安装
3.原生API操作Zookeeper
4.zkClient框架操作Zookeeper
5.Curator框架操作Zookeeper
原生API操作Zookeeper
- 首先要使用 java 操作 zookeeper, zookeeper 的 javaclient 使我们更轻松的去对 zookeeper 进行各种操作，我们引入 zookeeper-3.3.4. Jar 和 zkclient-0.1. Jar 即可。
- zookeeper-3.3.4. Jar 为官方提供的 javaAPI, zkclient-0.1. Jar 则为在源生 api 基础之上进行扩展的开源 JAVA 客户端。
zookeeper-3.3.4. Jar API介绍
-
创建会话方法：客户端可以通过创建一个 zookeeper 实例来连接 zookeeper 服务器。
ZooKeeper zk = new ZooKeeper(); //后面会有样例详细介绍怎么使用
//Zookeeper类一共了 4 个构造方法，根据参数不同,参数说明如下：
//connectString：连接服务器列表，用”，“分割。
//sessionTimeout：心跳检测时间周期（毫秒）
//wather：事件处理通知器。
//canBeReadOnly：标识当前会话是否支持只读。
//sessionld 和 sessionPasswd：提供连接 zookeeper 的 sessionld 和密码，通过这俩个确定唯一一台客户端，目的是可以提供重复会话。
-
创建节点（znode）方法：create:
提供了两套创建节点的方法，同步和异步创建节点方式。
-
同步方式：
- 参数 1，节点路径（名称）: /nodeName（不允许递归创建节点，也就是说在父节点不存在的情况下，不允许创建子节点）
- 参数 2，节点内容：要求类型是字节数组（也就是说，不支持序列化方式，如果需要实现序列化，可使用 java 相关序列化框架，如 Hessian、Kryo 框架）
- 参数 3，节点权限：使用 Ids. OPEN_ ACL UNSAFE 开放权限即可。（这个参数一般在权限没有太高要求的场景下，没必要关注）
- 参数 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 的 ctx 值
- name：实际在服务器端创建节点的名称
- 参数 5，注册一个异步回调函数，要实现 AsynCallBack. StringCallBack 接口，重写processResult (int rc, String path, Object ctx, String name）方法，当节点创建完毕后执行此方法。
参数 6，传递给回调函数的参数，一般为上下文（Context）信息
-
删除节点：delete 方法（api 提供了两个接口：同步删除和异步删除方式）
- 同步方式：
- 参数1，节点名称/deletePath
- 参数 2，版本号，即表明本次删除操作是针对该数据的某个版本进行的操作。
- 异步方式：（也是在同步方法基础上增加两个参数,使用方式和 create 方法一致）
- 参数 3:，一个异步回调函数
- 参数 4: 用于传递上下文信息的对象。
- 同步方式：
注意：在 zookeeper 中，只允许删除叶子节点信息，也就是说如果当前节点不是叶子节点则无法删除，或必须先删除其下所有子节点。
-
样例程序(创建会话,创建节点,删除节点)
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
/**
* Zookeeper base学习笔记
*/
public class ZookeeperBase {
/** zookeeper地址 */
static final String CONNECT_ADDR = "192.168.0.160:2181,192.168.0.161:2181,192.168.0.162:2181";
/** session超时时间 */
static final int SESSION_OUTTIME = 2000;//ms
/** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
public static void main(String[] args) throws Exception{
ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
@Override
public void process(WatchedEvent event) {
//获取事件的状态
KeeperState keeperState = event.getState();
EventType eventType = event.getType();
//如果是建立连接
if(KeeperState.SyncConnected == keeperState){
if(EventType.None == eventType){
//如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
connectedSemaphore.countDown();
System.out.println("zk 建立连接");
}
}
}
});
//进行阻塞
connectedSemaphore.await();
System.out.println("..");
//创建父节点
// zk.create("/testRoot", "testRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//创建子节点
// zk.create("/testRoot/children", "children data".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//获取节点洗信息
// byte[] data = zk.getData("/testRoot", false, null);
// System.out.println(new String(data));
// System.out.println(zk.getChildren("/testRoot", false));
//修改节点的值
// zk.setData("/testRoot", "modify data root".getBytes(), -1);
// byte[] data = zk.getData("/testRoot", false, null);
// System.out.println(new String(data));
//判断节点是否存在
// System.out.println(zk.exists("/testRoot/children", false));
//删除节点
// zk.delete("/testRoot/children", -1);
// System.out.println(zk.exists("/testRoot/children", false));
zk.close();
}
}
-
GetChildren 读取数据方法：包括子节点列表的获取和子节点数据的获取。
- 参数 1，path：获取指定节点的下的数据（获取子节点列表）
- 参数 2，watcher：注册的 watcher，一旦在本次子节点获取后，子节点列表发生变更的话，那么就会向客户端发送通知。该参数允许为 null。(如果为null,但wath参数为true时,会去取实例化ZooKeeper()连接对象时设置的那个watcher)
- 参数 3，wath：表明是否需要注册一个 watcher；如果为 true，则会使用到 zookeeper 客户端上文中提到的那个默认 watcher。如果为 false，则表明不需要注册 Watcher。
- 参数 4，cb：回调函数。
- 参数 5，ctx：上下文信息对象。
- 参数 6，stat：指定数据节点的节点状态信息。
注意：当我们获取指定节点的子节点列表后，还需要订阅这个子节点列表的变化通知，这时候就可以通过注册一个 watcher 来实现，当子节点被添加或删除时，服务器端就会触发一个“NodeChildrenChanged“类型的事件通知，需要注意的是服务器端发送给客户端的事件通知中，是不包含最新的节点列表的，客户端必须主动从新进行获取，通常在客户端收到这个事件通知后，就可以再次主动获取最新的子节点列表了。也就是说，zookeeper 服务端在向客户端发送 watcher"NodeChildrenChanged“事件通知的时候，仅仅只发了一个通知，不会把节点变化情况发给客户端，需要客户端自己重新获取，另外 Watcher 通知是一次性的，即触发后失效，因此客户端需要反复注册 Watcher 才行。
-
getData 方法：获取指定节点的数据内容。
- 参数 1，path：路径
- 参数 2，watcher：注册的 watcher 对象。一旦之后节点内容有变更，则会像客户端发送通知，该参数允许为 null。
- 参数 3，stat：指定节点的状态信息。
- 参数 4，watch：是否使用 watcher，如果为 true 则使用默认上文中的 watcher, false 则不使用 watcher。
- 参数 5，cb：回调函数。
- 参数 6，ctx：用于传递的下文信息对象。
注意：该方法和 getChildren 方法基本相同，主要是注册的 watcher 有所不同，客户端在获取一个阶段数据内容时，是可以进行 watcher 注册的，一旦节点发生变更，则服务器端会发送给客户端一个“NodeDataChanged“的事件通知。
-
setData 方法：修改指定节点的数据内容。
- 参数 1, path：路径。
- 参数 2, data：数据内容。
- 参数 3，版本号（-1 覆盖之前所有的版本）
- 参数 4，cb：回调函数。
- 参数 5，ctx：用于传递的下文信息对象。
-
exists 方法：检测节点是否存在。
- 参数 1，path：路径
- 参数 2，watcher：注册的 watcher 对象。一旦之后节点内容有变更，则会像客户端发送通知，该参数允许为 null。（用于三类事件监听：节点的创建、删除、更新）
- 参数 3，watch：是否使用 watcher，如果为 true 则使用默认上文中的 watcher, false 则不使用 watcher。
- 参数 4，cb：回调函数。
- 参数 5，ctx：用于传递的下文信息对象。
注意：exists 方法意义在于无论节点是否存在，都可以进行注册 watcher，能够对节点的创建、删除和修改进行监听，但是其子节点发送各种变化，都不会通知客户端。
-
Zookeeper 有 watch 事件，是一次性触发的，当 watch 监视的数据发生变化时，通知设置了该 watch 的 client，即 watcher.
-
同样，其 watcher 是监听数据发送了某些变化，那就一定会有对应的事件类型，和状态类型。
- 事件类型： (znode 节点相关的）
- EventType. NodeCreated
- EventType. NodeDataChanged
- EventType. NodeChildrenChanged
- EventType. NodeDeleted
- 状态类型：（是跟客户端实例相关的）
- KeeperState. Disconnected
- KeeperState. SyncConnected
- KeeperState. AuthFailed
- KeeperState. Expired
- 事件类型： (znode 节点相关的）
-
Watcher 的特性：一次性、客户端串行执行、轻量。
- 一次性：对于 ZK 的 watcher，你只需要记住一~点：zookeeper 有 watch 事件，是一次性触发的，当 watch 监视的数据发生变化时，通知设置了该 watch 的 client，即 watcher，由于 zookeeper 的监控都是一次性的所以每次必须设置监控。
- 客户端串行执行：客户端 Watcher 回调的过程是一个串行同步的过程，这为我们保证了顺序，同时需要开发人员注意一点，千万不要因为一个 Watcher 的处理逻辑影响了整个客户端的 Watcher 回调。
- 轻量：WatchedEvent 是 Zookeeper 整个 Watcher 通知机制的最小通知单元，整个结构只包含三部分：通知状态、事件类型和节点路径。也就是说 Watcher 通知非常的简单，只会告诉客户端发生了事件而不会告知其具体内容，需要客户自己去进行获取，比如 NodeDataChanged 事件，Zookeeper 只会通知客户端指定节点的数据发生了变更，而不会直接提供具体的数据内容。
我们通过一个示例，详细学习下 Watcher 的概念和其目的。Watcher 示例：
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.zookeeper.CreateMode;
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;
/**
* Zookeeper Wathcher
* 本类就是一个Watcher类（实现了org.apache.zookeeper.Watcher类）
*/
public class ZooKeeperWatcher implements Watcher {
/** 定义原子变量 */
AtomicInteger seq = new AtomicInteger();
/** 定义session失效时间 */
private static final int SESSION_TIMEOUT = 10000;
/** zookeeper服务器地址 */
private static final String CONNECTION_ADDR = "192.168.0.160:2181";
/** zk父路径设置 */
private static final String PARENT_PATH = "/testWatch";
/** zk子路径设置 */
private static final String CHILDREN_PATH = "/testWatch/children";
/** 进入标识 */
private static final String LOG_PREFIX_OF_MAIN = "【Main】";
/** zk变量 */
private ZooKeeper zk = null;
/** 信号量设置，用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
/**
* 创建ZK连接
* @param connectAddr ZK服务器地址列表
* @param sessionTimeout Session超时时间
*/
public void createConnection(String connectAddr, int sessionTimeout) {
this.releaseConnection();
try {
zk = new ZooKeeper(connectAddr, sessionTimeout, this);
System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
connectedSemaphore.await();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 关闭ZK连接
*/
public void releaseConnection() {
if (this.zk != null) {
try {
this.zk.close();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 创建节点
* @param path 节点路径
* @param data 数据内容
* @return
*/
public boolean createPath(String path, String data) {
try {
//设置监控(由于zookeeper的监控都是一次性的所以 每次必须设置监控)
this.zk.exists(path, true);
System.out.println(LOG_PREFIX_OF_MAIN + "节点创建成功, Path: " +
this.zk.create( /**路径*/
path,
/**数据*/
data.getBytes(),
/**所有可见*/
Ids.OPEN_ACL_UNSAFE,
/**永久存储*/
CreateMode.PERSISTENT ) +
", content: " + data);
} catch (Exception e) {
e.printStackTrace();
return false;
}
return true;
}
/**
* 读取指定节点数据内容
* @param path 节点路径
* @return
*/
public String readData(String path, boolean needWatch) {
try {
return new String(this.zk.getData(path, needWatch, null));
} catch (Exception e) {
e.printStackTrace();
return "";
}
}
/**
* 更新指定节点数据内容
* @param path 节点路径
* @param data 数据内容
* @return
*/
public boolean writeData(String path, String data) {
try {
System.out.println(LOG_PREFIX_OF_MAIN + "更新数据成功，path：" + path + ", stat: " +
this.zk.setData(path, data.getBytes(), -1));
} catch (Exception e) {
e.printStackTrace();
}
return false;
}
/**
* 删除指定节点
*
* @param path
* 节点path
*/
public void deleteNode(String path) {
try {
this.zk.delete(path, -1);
System.out.println(LOG_PREFIX_OF_MAIN + "删除节点成功，path：" + path);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 判断指定节点是否存在
* @param path 节点路径
*/
public Stat exists(String path, boolean needWatch) {
try {
return this.zk.exists(path, needWatch);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 获取子节点
* @param path 节点路径
*/
private List<String> getChildren(String path, boolean needWatch) {
try {
return this.zk.getChildren(path, needWatch);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 删除所有节点
*/
public void deleteAllTestPath() {
if(this.exists(CHILDREN_PATH, false) != null){
this.deleteNode(CHILDREN_PATH);
}
if(this.exists(PARENT_PATH, false) != null){
this.deleteNode(PARENT_PATH);
}
}
/**
* 收到来自Server的Watcher通知后的处理。
*/
@Override
public void process(WatchedEvent event) {
System.out.println("进入 process 。。。。。event = " + event);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (event == null) {
return;
}
// 连接状态
KeeperState keeperState = event.getState();
// 事件类型
EventType eventType = event.getType();
// 受影响的path
String path = event.getPath();
String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";
System.out.println(logPrefix + "收到Watcher通知");
System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
System.out.println(logPrefix + "事件类型:\t" + eventType.toString());
if (KeeperState.SyncConnected == keeperState) {
// 成功连接上ZK服务器
if (EventType.None == eventType) {
System.out.println(logPrefix + "成功连接上ZK服务器");
connectedSemaphore.countDown();
}
//创建节点
else if (EventType.NodeCreated == eventType) {
System.out.println(logPrefix + "节点创建");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.exists(path, true);
}
//更新节点
else if (EventType.NodeDataChanged == eventType) {
System.out.println(logPrefix + "节点数据更新");
System.out.println("我看看走不走这里........");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(logPrefix + "数据内容: " + this.readData(PARENT_PATH, true));
}
//更新子节点
else if (EventType.NodeChildrenChanged == eventType) {
System.out.println(logPrefix + "子节点变更");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(logPrefix + "子节点列表：" + this.getChildren(PARENT_PATH, true));
}
//删除节点
else if (EventType.NodeDeleted == eventType) {
System.out.println(logPrefix + "节点 " + path + " 被删除");
}
else ;
}
else if (KeeperState.Disconnected == keeperState) {
System.out.println(logPrefix + "与ZK服务器断开连接");
}
else if (KeeperState.AuthFailed == keeperState) {
System.out.println(logPrefix + "权限检查失败");
}
else if (KeeperState.Expired == keeperState) {
System.out.println(logPrefix + "会话失效");
}
else ;
System.out.println("--------------------------------------------");
}
/**
* <B>方法名称：</B>测试zookeeper监控<BR>
* <B>概要说明：</B>主要测试watch功能<BR>
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
//建立watcher
ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
//创建连接
zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
//System.out.println(zkWatch.zk.toString());
Thread.sleep(1000);
// 清理节点
zkWatch.deleteAllTestPath();
if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "")) {
Thread.sleep(1000);
// 读取数据
System.out.println("---------------------- read parent ----------------------------");
//zkWatch.readData(PARENT_PATH, true);
// 读取子节点
System.out.println("---------------------- read children path ----------------------------");
zkWatch.getChildren(PARENT_PATH, true);
// 更新数据
zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");
Thread.sleep(1000);
// 创建子节点
zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "");
Thread.sleep(1000);
zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");
}
Thread.sleep(50000);
// 清理节点
zkWatch.deleteAllTestPath();
Thread.sleep(1000);
zkWatch.releaseConnection();
}
}
我们再通过一个示例，模拟分布式情况下修改节点数据时,多个服务监听此节点,示例：
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
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.ZooKeeper;
public class ZKWatcher implements Watcher {
/** zk变量 */
private ZooKeeper zk = null;
/** 父节点path */
static final String PARENT_PATH = "/super";
/** 信号量设置，用于等待zookeeper连接建立之后 通知阻塞程序继续向下执行 */
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
private List<String> cowaList = new CopyOnWriteArrayList<String>();
/** zookeeper服务器地址 */
public static final String CONNECTION_ADDR = "192.168.0.160:2181,192.168.0.161:2181,192.168.0.162:2181";
/** 定义session失效时间 */
public static final int SESSION_TIMEOUT = 30000;
public ZKWatcher() throws Exception{
zk = new ZooKeeper(CONNECTION_ADDR, SESSION_TIMEOUT, this);
System.out.println("开始连接ZK服务器");
connectedSemaphore.await();
}
@Override
public void process(WatchedEvent event) {
// 连接状态
KeeperState keeperState = event.getState();
// 事件类型
EventType eventType = event.getType();
// 受影响的path
String path = event.getPath();
System.out.println("受影响的path : " + path);
if (KeeperState.SyncConnected == keeperState) {
// 成功连接上ZK服务器
if (EventType.None == eventType) {
System.out.println("成功连接上ZK服务器");
connectedSemaphore.countDown();
try {
if(this.zk.exists(PARENT_PATH, false) == null){
this.zk.create(PARENT_PATH, "root".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
List<String> paths = this.zk.getChildren(PARENT_PATH, true);
for (String p : paths) {
System.out.println(p);
this.zk.exists(PARENT_PATH + "/" + p, true);
}
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
//创建节点
else if (EventType.NodeCreated == eventType) {
System.out.println("节点创建");
try {
this.zk.exists(path, true);
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
//更新节点
else if (EventType.NodeDataChanged == eventType) {
System.out.println("节点数据更新");
try {
//update nodes call function
this.zk.exists(path, true);
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
//更新子节点
else if (EventType.NodeChildrenChanged == eventType) {
System.out.println("子节点 ... 变更");
try {
List<String> paths = this.zk.getChildren(path, true);
if(paths.size() >= cowaList.size()){
paths.removeAll(cowaList);
for(String p : paths){
this.zk.exists(path + "/" + p, true);
//this.zk.getChildren(path + "/" + p, true);
System.out.println("这个是新增的子节点 : " + path + "/" + p);
//add new nodes call function
}
cowaList.addAll(paths);
} else {
cowaList = paths;
}
System.out.println("cowaList: " + cowaList.toString());
System.out.println("paths: " + paths.toString());
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
//删除节点
else if (EventType.NodeDeleted == eventType) {
System.out.println("节点 " + path + " 被删除");
try {
//delete nodes call function
this.zk.exists(path, true);
} catch (KeeperException | InterruptedException e) {
e.printStackTrace();
}
}
else ;
}
else if (KeeperState.Disconnected == keeperState) {
System.out.println("与ZK服务器断开连接");
}
else if (KeeperState.AuthFailed == keeperState) {
System.out.println("权限检查失败");
}
else if (KeeperState.Expired == keeperState) {
System.out.println("会话失效");
}
else ;
System.out.println("--------------------------------------------");
}
}
-
实例两个上面ZKWatcher对象,ZKWatcher其实就是简单了实现了监听了/super数据节点,你可以把这想象成两个不同的服务
public class Client1 {
public static void main(String[] args) throws Exception{
ZKWatcher myWatcher = new ZKWatcher();
Thread.sleep(100000000);
}
}
public class Client2 {
public static void main(String[] args) throws Exception{
ZKWatcher myWatcher = new ZKWatcher();
Thread.sleep(100000000);
}
}
-
通过下面这个小测试,测试对/super数据节点不同的操作,可以看到上面两个小程序会分别打印对应的信息
import java.util.concurrent.CountDownLatch;
import org.apache.zookeeper.CreateMode;
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;
public class Test {
/** zookeeper地址 */
static final String CONNECT_ADDR = "192.168.1.106:2181,192.168.1.107:2181,192.168.1.108:2181";
/** session超时时间 */
static final int SESSION_OUTTIME = 2000;//ms
/** 信号量，阻塞程序执行，用于等待zookeeper连接成功，发送成功信号 */
static final CountDownLatch connectedSemaphore = new CountDownLatch(1);
public static void main(String[] args) throws Exception{
ZooKeeper zk = new ZooKeeper(CONNECT_ADDR, SESSION_OUTTIME, new Watcher(){
@Override
public void process(WatchedEvent event) {
//获取事件的状态
KeeperState keeperState = event.getState();
EventType eventType = event.getType();
//如果是建立连接
if(KeeperState.SyncConnected == keeperState){
if(EventType.None == eventType){
//如果建立连接成功，则发送信号量，让后续阻塞程序向下执行
connectedSemaphore.countDown();
System.out.println("zk 建立连接");
}
}
}
});
//进行阻塞
connectedSemaphore.await();
// //创建子节点
// zk.create("/super/c1", "c1".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//创建子节点
// zk.create("/super/c2", "c2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//创建子节点
zk.create("/super/c3", "c3".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//创建子节点
// zk.create("/super/c4", "c4".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
// zk.create("/super/c4/c44", "c44".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//获取节点信息
// byte[] data = zk.getData("/testRoot", false, null);
// System.out.println(new String(data));
// System.out.println(zk.getChildren("/testRoot", false));
//修改节点的值
// zk.setData("/super/c1", "modify c1".getBytes(), -1);
// zk.setData("/super/c2", "modify c2".getBytes(), -1);
// byte[] data = zk.getData("/super/c2", false, null);
// System.out.println(new String(data));
// //判断节点是否存在
// System.out.println(zk.exists("/super/c3", false));
// //删除节点
// zk.delete("/super/c3", -1);
zk.close();
}
}
-
Zookeeperd的ACL
-
ACL (Access ControlList），Zookeeper 作为一个分布式协调框架，其内部存储的都是一些关乎分布式系统运行时状态的元数据，尤其是设计到一些分布式锁、Master 选举和协调等应用场景。我们需要有效地保障 zookeeper 中的数据安全，Zookeeper 提供一套完善的 ACL 权限控制机制来保障数据的安全。ZK 提供了三种模式。权限模式、授权对象、权限。
- 权限模式：Scheme，开发人员最多使用的如下四种权限模式：
- IP: ip 模式通过 ip 地址粒度来进行控制权限，例如配置了：ip:192.168.1.107 即表示权限控制都是针对这个 ip 地址的，同时也支持按网段分配，比如 192.168.1. *
- Digest: digest 是最常用的权限控制模式，也更符合我们对权限控制的认识，其类似于_ "usermame: password“形式的权限标识进行权限配置。ZK 会对形成的权限标识先后进行俩次编码处理，分别是 SHA-1 加密算法、BASE64 编码。
- World: World 是一直最开放的权限控制模式。这种模式可以看做为特殊的 Digest，他仅仅是一个标识而已。
- Super：超级用户模式，在超级用户模式下可以对 ZK 任意进行操作。
- 权限模式：Scheme，开发人员最多使用的如下四种权限模式：
授权对象：指的是权限赋予的用户或者一个指定的实体，例如 ip 地址或机器等。在不同的模式下，授权对象是不同的。这种模式和权限对象一一对应。
-
权限：权限就是指那些通过权限检测后可以被允许执行的操作，在 ZK 中，对数据的操作权限分为头下五大类：
- CREATE、DELETE、READ、WRITE、ADMIN
-
我们通过一个示例，详细学习下 Auth 的概念和其目的。Auth 示例： [ZooKeeperAuth]
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.zookeeper.CreateMode;
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.ACL;
import org.apache.zookeeper.data.Stat;
/**
* Zookeeper 节点授权
*/
public class ZookeeperAuth implements Watcher {
/** 连接地址 */
final static String CONNECT_ADDR = "192.168.0.160:2181";
/** 测试路径 */
final static String PATH = "/testAuth";
final static String PATH_DEL = "/testAuth/delNode";
/** 认证类型 */
final static String authentication_type = "digest";
/** 认证正确方法 */
final static String correctAuthentication = "123456";
/** 认证错误方法 */
final static String badAuthentication = "654321";
static ZooKeeper zk = null;
/** 计时器 */
AtomicInteger seq = new AtomicInteger();
/** 标识 */
private static final String LOG_PREFIX_OF_MAIN = "【Main】";
private CountDownLatch connectedSemaphore = new CountDownLatch(1);
@Override
public void process(WatchedEvent event) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (event==null) {
return;
}
// 连接状态
KeeperState keeperState = event.getState();
// 事件类型
EventType eventType = event.getType();
// 受影响的path
String path = event.getPath();
String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";
System.out.println(logPrefix + "收到Watcher通知");
System.out.println(logPrefix + "连接状态:\t" + keeperState.toString());
System.out.println(logPrefix + "事件类型:\t" + eventType.toString());
if (KeeperState.SyncConnected == keeperState) {
// 成功连接上ZK服务器
if (EventType.None == eventType) {
System.out.println(logPrefix + "成功连接上ZK服务器");
connectedSemaphore.countDown();
}
} else if (KeeperState.Disconnected == keeperState) {
System.out.println(logPrefix + "与ZK服务器断开连接");
} else if (KeeperState.AuthFailed == keeperState) {
System.out.println(logPrefix + "权限检查失败");
} else if (KeeperState.Expired == keeperState) {
System.out.println(logPrefix + "会话失效");
}
System.out.println("--------------------------------------------");
}
/**
* 创建ZK连接
*
* @param connectString
* ZK服务器地址列表
* @param sessionTimeout
* Session超时时间
*/
public void createConnection(String connectString, int sessionTimeout) {
this.releaseConnection();
try {
zk = new ZooKeeper(connectString, sessionTimeout, this);
//添加节点授权
zk.addAuthInfo(authentication_type,correctAuthentication.getBytes());
System.out.println(LOG_PREFIX_OF_MAIN + "开始连接ZK服务器");
//倒数等待
connectedSemaphore.await();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 关闭ZK连接
*/
public void releaseConnection() {
if (this.zk!=null) {
try {
this.zk.close();
} catch (InterruptedException e) {
}
}
}
/**
*
* <B>方法名称：</B>测试函数<BR>
* <B>概要说明：</B>测试认证<BR>
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
ZookeeperAuth testAuth = new ZookeeperAuth();
testAuth.createConnection(CONNECT_ADDR,2000);
List<ACL> acls = new ArrayList<ACL>(1);
for (ACL ids_acl : Ids.CREATOR_ALL_ACL) {
acls.add(ids_acl);
}
try {
zk.create(PATH, "init content".getBytes(), acls, CreateMode.PERSISTENT);
System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH + ", 初始内容是: init content");
} catch (Exception e) {
e.printStackTrace();
}
try {
zk.create(PATH_DEL, "will be deleted! ".getBytes(), acls, CreateMode.PERSISTENT);
System.out.println("使用授权key：" + correctAuthentication + "创建节点："+ PATH_DEL + ", 初始内容是: init content");
} catch (Exception e) {
e.printStackTrace();
}
// 获取数据
getDataByNoAuthentication();
getDataByBadAuthentication();
getDataByCorrectAuthentication();
// 更新数据
updateDataByNoAuthentication();
updateDataByBadAuthentication();
updateDataByCorrectAuthentication();
// 删除数据
deleteNodeByBadAuthentication();
deleteNodeByNoAuthentication();
deleteNodeByCorrectAuthentication();
//
Thread.sleep(1000);
deleteParent();
//释放连接
testAuth.releaseConnection();
}
/** 获取数据：采用错误的密码 */
static void getDataByBadAuthentication() {
String prefix = "[使用错误的授权信息]";
try {
ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
//授权
badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
Thread.sleep(2000);
System.out.println(prefix + "获取数据：" + PATH);
System.out.println(prefix + "成功获取数据：" + badzk.getData(PATH, false, null));
} catch (Exception e) {
System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
}
}
/** 获取数据：不采用密码 */
static void getDataByNoAuthentication() {
String prefix = "[不使用任何授权信息]";
try {
System.out.println(prefix + "获取数据：" + PATH);
ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
Thread.sleep(2000);
System.out.println(prefix + "成功获取数据：" + nozk.getData(PATH, false, null));
} catch (Exception e) {
System.err.println(prefix + "获取数据失败，原因：" + e.getMessage());
}
}
/** 采用正确的密码 */
static void getDataByCorrectAuthentication() {
String prefix = "[使用正确的授权信息]";
try {
System.out.println(prefix + "获取数据：" + PATH);
System.out.println(prefix + "成功获取数据：" + zk.getData(PATH, false, null));
} catch (Exception e) {
System.out.println(prefix + "获取数据失败，原因：" + e.getMessage());
}
}
/**
* 更新数据：不采用密码
*/
static void updateDataByNoAuthentication() {
String prefix = "[不使用任何授权信息]";
System.out.println(prefix + "更新数据： " + PATH);
try {
ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
Thread.sleep(2000);
Stat stat = nozk.exists(PATH, false);
if (stat!=null) {
nozk.setData(PATH, prefix.getBytes(), -1);
System.out.println(prefix + "更新成功");
}
} catch (Exception e) {
System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
}
}
/**
* 更新数据：采用错误的密码
*/
static void updateDataByBadAuthentication() {
String prefix = "[使用错误的授权信息]";
System.out.println(prefix + "更新数据：" + PATH);
try {
ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
//授权
badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
Thread.sleep(2000);
Stat stat = badzk.exists(PATH, false);
if (stat!=null) {
badzk.setData(PATH, prefix.getBytes(), -1);
System.out.println(prefix + "更新成功");
}
} catch (Exception e) {
System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
}
}
/**
* 更新数据：采用正确的密码
*/
static void updateDataByCorrectAuthentication() {
String prefix = "[使用正确的授权信息]";
System.out.println(prefix + "更新数据：" + PATH);
try {
Stat stat = zk.exists(PATH, false);
if (stat!=null) {
zk.setData(PATH, prefix.getBytes(), -1);
System.out.println(prefix + "更新成功");
}
} catch (Exception e) {
System.err.println(prefix + "更新失败，原因是：" + e.getMessage());
}
}
/**
* 不使用密码 删除节点
*/
static void deleteNodeByNoAuthentication() throws Exception {
String prefix = "[不使用任何授权信息]";
try {
System.out.println(prefix + "删除节点：" + PATH_DEL);
ZooKeeper nozk = new ZooKeeper(CONNECT_ADDR, 2000, null);
Thread.sleep(2000);
Stat stat = nozk.exists(PATH_DEL, false);
if (stat!=null) {
nozk.delete(PATH_DEL,-1);
System.out.println(prefix + "删除成功");
}
} catch (Exception e) {
System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
}
}
/**
* 采用错误的密码删除节点
*/
static void deleteNodeByBadAuthentication() throws Exception {
String prefix = "[使用错误的授权信息]";
try {
System.out.println(prefix + "删除节点：" + PATH_DEL);
ZooKeeper badzk = new ZooKeeper(CONNECT_ADDR, 2000, null);
//授权
badzk.addAuthInfo(authentication_type,badAuthentication.getBytes());
Thread.sleep(2000);
Stat stat = badzk.exists(PATH_DEL, false);
if (stat!=null) {
badzk.delete(PATH_DEL, -1);
System.out.println(prefix + "删除成功");
}
} catch (Exception e) {
System.err.println(prefix + "删除失败，原因是：" + e.getMessage());
}
}
/**
* 使用正确的密码删除节点
*/
static void deleteNodeByCorrectAuthentication() throws Exception {
String prefix = "[使用正确的授权信息]";
try {
System.out.println(prefix + "删除节点：" + PATH_DEL);
Stat stat = zk.exists(PATH_DEL, false);
if (stat!=null) {
zk.delete(PATH_DEL, -1);
System.out.println(prefix + "删除成功");
}
} catch (Exception e) {
System.out.println(prefix + "删除失败，原因是：" + e.getMessage());
}
}
/**
* 使用正确的密码删除节点
*/
static void deleteParent() throws Exception {
try {
Stat stat = zk.exists(PATH_DEL, false);
if (stat == null) {
zk.delete(PATH, -1);
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
