Watcher ZK狀態 事件型別

zookeeper有Watch事件，是一次性出發的，當watch 監視的資料發生變化時,通知設定了該watch的client，即Watcher

同樣，其Watcher事監聽資料傳送了某些變化，那就一定會有對應的事件型別和狀態型別

事件型別：（znode節點相關）

EventType.NodeCreated

EventType.NodeDateChanged

EventType.NodeChildrenchanged

EventType.NodeDeleted

狀態型別：（是跟客戶端例項相關的） Flow 與 Leader Watcher 

KeeperState:Disconnected

KeeperState:SyncConnected

KeeperState:AuthFailed

KeeperState:Expired

c1 c2 相當於client 通過建立一個執行緒watch 事件監聽 zookeeper節點的變化 所以C1 C2 相當於 Watcher ，而且client端可以有多個Watcher 一個Watcher監控一個節點，但Watch事件，是一次性出發的，如果一直監控節點的話 要重複的進行watch。

下面我們通過程式碼來看一下：

/**
 * 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.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();
	}

}