Zookeeper系列(3)：使用Java操作Zookeeper

簡介

Java操作Zookeeper有很多種方式，如zookeeper、zkclient、curator等等，下面介紹下使用zkclient的方式操作Zookeeper。

Maven依賴：

<dependency>
	<groupId>com.101tec</groupId>
	<artifactId>zkclient</artifactId>
	<version>0.11</version>
</dependency>

使用zkclient操作Zookeeper

創建節點：

@Test
public void testCreateNode() {
	//建立連接
	//zkServers: Zookeeper服務器IP地址和端口號，如果是集群情況下用逗號分割多個Zookeeper服務器地址
	//sessionTimeout: 會話超時時間
	//connectionTimeout: 連接超時時間
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);

	//創建數據
	Student student = new Student();
	student.setName("張三");
	student.setAge(18);
	student.setPhone("1585454xxxx");

	//創建持久節點
	zkClient.createPersistent("/p_node", student);
	//創建持久順序節點
	zkClient.createPersistentSequential("/ps_node", student);
	//創建臨時節點
	zkClient.createEphemeral("/e_node", student);
	//創建臨時順序節點
	zkClient.createEphemeralSequential("/ps_node", student);

	//關閉客戶端
	//關閉客戶端的同時，前面創建的臨時節點也會被刪除
	zkClient.close();
}

讀取節點數據：

@Test
public void testReadNodeData() {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);

	Stat stat = new Stat();
	Student student = zkClient.readData("/p_node", stat);
	System.out.println("節點狀態信息：" + JSON.toJSONString(student));
	System.out.println("節點數據：" + JSON.toJSONString(stat));

	zkClient.close();
}
//輸出結果：
//節點狀態信息：{"age":18,"name":"張三","phone":"1585454xxxx"}
//節點數據：{"aversion":0,"ctime":1619165355431,"cversion":0,"czxid":165,"dataLength":260,"ephemeralOwner":0,"mtime":1619165355431,"mzxid":165,"numChildren":0,"pzxid":165,"version":0}

刪除節點：

@Test
public void testDeleteNode() {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);

	//刪除節點，不能刪除帶有子節點的節點
	zkClient.delete("/p_node");
	//刪除節點，遞歸刪除所有子孫節點
	zkClient.deleteRecursive("/p_node2");

	zkClient.close();
}

更新節點數據：

@Test
public void testWriteNodeData() {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);

	//更新節點數據
	zkClient.writeData("/p_node", "myData1");
	//使用CAS更新節點數據
	//zkClient.writeData("/p_node", "myData2", 1);

	zkClient.close();
}

獲取子節點列表：

@Test
public void testGetChildNodes() {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);

	//獲取子節點列表
	List<String> childList = zkClient.getChildren("/p_node");
	childList.stream().forEach(System.out::println);

	zkClient.close();
}

節點監聽

Java節點監聽都是永久的，觸發一次后不會被刪除。

監聽節點的子節點變化：

@Test
public void testSubscribeChildChanges() throws IOException {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);
	zkClient.subscribeChildChanges("/p_node", new IZkChildListener(){

		//子節點改變時調用
		@Override
		public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
			System.out.println(parentPath + "子節點發生改變");
			System.out.println("當前子節點列表：" + currentChilds);
		}

	});

	//阻塞客戶端，便於測試
	System.in.read();
}

監聽節點數據變化：

@Test
public void testSubscribeDataChanges() throws IOException {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);
	zkClient.subscribeDataChanges("/p_node", new IZkDataListener(){

		//節點數據改變時調用
		@Override
		public void handleDataChange(String dataPath, Object data) throws Exception {
			System.out.println(dataPath + "節點數據發生變化");
			System.out.println("修改后的數據為：" + data.toString());
		}

		//節點被刪除時調用
		@Override
		public void handleDataDeleted(String dataPath) throws Exception {
			System.out.println(dataPath + "節點已被刪除");
		}
	});

	//阻塞客戶端，便於測試
	System.in.read();
}

監聽Zookeeper連接狀態變化：

@Test
public void testSubscribeStateChanges() throws IOException {
	ZkClient zkClient = new ZkClient("127.0.0.1:2181", 60000, 5000);
	zkClient.subscribeStateChanges(new IZkStateListener(){

		//當zookeeper連接狀態改變時調用
		@Override
		public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
			System.out.println("當前狀態" + state);
		}

		//在zookeeper會話過期並且創建了一個新的會話之后調用
		@Override
		public void handleNewSession() throws Exception {
			System.out.println("會話過期，已創建新的會話");
		}

		//當會話不能重新建立時調用
		@Override
		public void handleSessionEstablishmentError(Throwable error) throws Exception {
			error.printStackTrace();
		}
	});

	//阻塞客戶端，便於測試
	System.in.read();
}

刪除節點監聽可以調用以下方法：

刪除所有節點監聽：unsubscribeAll()

刪除指定的節點監聽：unsubscribeChildChanges()，unsubscribeDataChanges()，unsubscribeStateChanges()