## 一、Watcher 、ZK狀態、事件類型
zookeeper有watch事件，是一次性觸發的，當watch監視的數據發生變化時，通知設置了該watch的client的watch
同樣，其watch是監聽數據發生了某些變法，那就一定會有對應的時間類型和狀態類型
事件類型：（znode節點想過的）
EventType.NodeCreated 創建
EventType.NodeDataChanged 節點的數據發生變更
EventType.NodeChildrenChanged 子節點發生變更
EventType.NodeDeteted
狀態類型 （是跟客戶端實例相關的）
KeeperState.Disconnected 連接不上
KeeperState.SyncConnected 連接上了
KeeperState.AuthFailed 認證失敗
KeeperState.Expired 過期
## 二、 代碼的實現
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.80.88: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();
}
}
## 三、總結
zookeeper 是如何監聽到節點的變法。他有watch的事件，通過watch觀察到節點的變法，然后觸發process事件
然后檢查數據類型，在檢測狀態類的節點變法。
具體怎么做：
第一步：注冊watch事件，使用zookeeper里面擁有該事件，什么時候觸發
第二步：需要想節點的注冊是否啟動事件 意思就是是否觸發process方法
在這里需要注意：
監聽只能監聽到已經子節點的變法，如果想法在子節點發生改變的時候，那么父節點需要啟動事件
每次監聽都要開啟是否啟用事件。上面有詳細的代碼過程。
