ZooKeeper watch机制核心讲解

ZooKeeper watch机制核心讲解

https://blog.csdn.net/Future_LL/article/details/87700481

 

Zookeeper中全部读操做（getData()，getChildren()，exists()）均可以设置Watch选项。Watch事件具备one-time trigger（一次性触发）的特性，若是Watch监视的Znode有变化，那么就会通知设置该Watch的客户端。

客户端在Znode设置了Watch时，若是Znode内容发生改变，那么客户端就会得到Watch事件。例如：客户端设置getData("/znode1", true)后，若是/znode1发生改变或者删除，那么客户端就会获得一个/znode1的Watch事件，可是/znode1再次发生变化，那客户端是没法收到Watch事件的，除非客户端设置了新的Watch。

Watch
ZooKeeper有watch事件，是一次性触发的【每次数据要发生变化以前都要手动建立watch】，当watch监视的数据发生时，通知设置了该watch的client，客户端即watcher。一样，其watcher是监听数据发送了某些变化，那就必定会有对应的事件类型和状态类型，一个客户端能够监控多个节点，在代码中体如今new了几个就产生几个watcher，只要节点变化都要执行一边process
事件类型：（znode节点相关的）【针对的是你所观察的一个节点而言的】
EventType.NodeCreated 【节点建立】
EventType.NodeDataChanged 【节点数据发生变化】
EventType.NodeChildrenChanged 【这个节点的子节点发生变化】
EventType.NodeDeleted 【删除当前节点】
状态类型：（是跟客户端实例相关的）【ZooKeeper集群跟应用服务之间的状态的变动】
KeeperState.Disconnected 【没有链接上】
KeeperState.SyncConnected 【链接上】
KeeperState.AuthFailed 【认证失败】
KeeperState.Expired 【过时】
watcher的特性：一次性，客户端串行执行，轻量
一次性：对于ZooKeeper的watcher，你只须要记住一点，ZooKeeper有watch事件，是一次性触发的，当watch监视的数据发生变化时，通知设置该watch的client，即watcher，因为ZooKeeper的监控都是一次性的，因此每次必须设置监控
客户端串行执行：客户端watcher回调的过程是一个串行同步的过程，这为咱们保证了顺序，同时须要开发人员注意一点，千万不要由于一个watcher的处理逻辑影响了整个客户端的watcher回调
轻量：WatchedEvent是ZooKeeper整个Watcher通知机制的最小通知单元，整个结构只包含三个部分：通知状态、事件类型和节点路径。也就是说Watcher通知很是的简单，只会告知客户端发生了事件而不会告知其具体内容，须要客户端本身去进行获取，好比NodeDataChanged事件，ZooKeeper只会通知客户端指定节点的数据发生了变动，而不会直接提供具体的数据内容
package bhz.zookeeper.watcher;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import javafx.scene.shape.Path;
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类）
* @author（alienware）
* @since 2019-2-19
*/
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.1.24:2181";
/** zk父路径设置 */
private static final String PARENT_PATH = "/p";
/** zk子路径设置 */
private static final String CHILDREN_PATH = "/p/c";
/** 进入标识 */
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 {
//this表示把当前对象进行传递到其中去（也就是在主函数里实例化的new ZooKeeperWatcher()实例对象）
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, boolean needWatch) {
try {
//设置监控(因为zookeeper的监控都是一次性的因此 每次必须设置监控)
this.zk.exists(path, needWatch);
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 {
System.out.println("读取数据操做...");
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;
}
return true;
}

/**
* 删除指定节点
*
* @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 {
System.out.println("读取子节点操做...");
return this.zk.getChildren(path, needWatch);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}

/**
* 删除全部节点
*/
public void deleteAllTestPath(boolean needWatch) {
if(this.exists(CHILDREN_PATH, needWatch) != null){
this.deleteNode(CHILDREN_PATH);
}
if(this.exists(PARENT_PATH, needWatch) != 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();
// 原子对象seq 记录进入process的次数
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();
}
}
//更新节点
else if (EventType.NodeDataChanged == eventType) {
System.out.println(logPrefix + "节点数据更新");
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//更新子节点
else if (EventType.NodeChildrenChanged == eventType) {
System.out.println(logPrefix + "子节点变动");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//删除节点
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 //当前客户端能够称为一个watcher 观察者角色
ZooKeeperWatcher zkWatch = new ZooKeeperWatcher();
//建立链接
zkWatch.createConnection(CONNECTION_ADDR, SESSION_TIMEOUT);
//System.out.println(zkWatch.zk.toString());

Thread.sleep(1000);

// 清理节点
// zkWatch.deleteAllTestPath(false);

//-----------------第一步: 建立父节点 /p ------------------------//
// 建立父节点
/**
* data:节点的内容是当前的时间毫秒值，needWATCH:是否须要监控
*/
if (zkWatch.createPath(PARENT_PATH, System.currentTimeMillis() + "", true)) {

Thread.sleep(1000);

//-----------------第二步: 读取节点 /p 和 读取/p节点下的子节点(getChildren)的区别 --------------//
// 读取数据
// zkWatch.readData(PARENT_PATH, true);
// 也能够这样作，表示当前的节点须要继续watch，更新数据操做就会被监控到
zkWatch.exists(PARENT_PATH,true);

// 读取子节点(监控childNodeChange事件)
// 这里的监听仍是监听PARENT_PATH，只是这里多了一个关于子节点的触发，这里的触发仍是父节点
// 只要子节点发生变化【删除、修改、增长】那么就触发，只会触发:NodeChildrenChanged
zkWatch.getChildren(PARENT_PATH, true);

// 更新数据，由于上边watch，因此更新操做会被监听到，当更新结束，上述的watch关系会消失
zkWatch.writeData(PARENT_PATH, System.currentTimeMillis() + "");

Thread.sleep(1000);
// 建立子节点
// 若是needWatch为false，只是单纯的建立了一个子节点，没有触发任何watch:NodeCreated
zkWatch.createPath(CHILDREN_PATH, System.currentTimeMillis() + "", true);


//-----------------第三步: 创建子节点的触发 --------------//
// zkWatch.getChildren(CHILDREN_PATH, true);
// zkWatch.createPath(CHILDREN_PATH + "/c1", System.currentTimeMillis() + "", true);
// zkWatch.getChildren(CHILDREN_PATH + "/c1", true);
// zkWatch.createPath(CHILDREN_PATH + "/c1/c2", System.currentTimeMillis() + "", true);

//-----------------第四步: 更新子节点数据的触发 --------------//
//在进行修改以前，咱们须要watch一下这个节点：
Thread.sleep(1000);
zkWatch.readData(CHILDREN_PATH, true);
zkWatch.writeData(CHILDREN_PATH, System.currentTimeMillis() + "");

}

Thread.sleep(10000);
// 清理节点
zkWatch.deleteAllTestPath(true);

Thread.sleep(10000);
zkWatch.releaseConnection();

}

---------------------
做者：Future_LL
来源：CSDN
原文：https://blog.csdn.net/Future_LL/article/details/87700481
版权声明：本文为博主原创文章，转载请附上博文连接！