zk使用通知移除節點
阿新 • • 發佈:2018-11-03
前面:https://www.cnblogs.com/toov5/p/9899238.html 服務發生宕機 咋辦?
發個事件通知,告知大家喲,
會有通知事件哦
看專案:
服務端:
package com.toov5.zkDubbo; import java.io.IOException; import java.net.ServerSocket; import java.net.Socket; import org.I0Itec.zkclient.ZkClient; //##ServerScoekt服務端 public class ZkServerScoekt implementsRunnable { private static int port = 18083; //socket 服務啟動後的所使用的 埠號 public static void main(String[] args) throws IOException { ZkServerScoekt server = new ZkServerScoekt(port); //建構函式傳入port regServer(); //去zk註冊 Thread thread = new Thread(server); thread.start(); }public ZkServerScoekt(int port) { this.port = port; } //註冊服務 public static void regServer(){ //1、建立zk連線 ZkClient zkClient = new ZkClient("192.168.91.5",5000,10000); //2.先建立父節點 String root = "/toov5"; if (!zkClient.exists(root)) { //如果父節點不存,直接建立父節點 zkClient.createPersistent(root); //持久節點 } //3、建立子節點 String nodeName = root+"/service_"+port; String nodeValue="127.0.0.1:"+port; if (zkClient.exists(nodeName)) { //如果存在 直接刪除掉 zkClient.delete(nodeName); } zkClient.createEphemeral(nodeName,nodeValue); //位元組的 臨時節點 如果服務宕機了 找不到了 System.out.println("服務註冊成功"+nodeName); } public void run() { ServerSocket serverSocket = null; try { serverSocket = new ServerSocket(port); System.out.println("Server start port:" + port); Socket socket = null; while (true) { socket = serverSocket.accept(); new Thread(new ServerHandler(socket)).start(); } } catch (Exception e) { e.printStackTrace(); } finally { try { if (serverSocket != null) { serverSocket.close(); } } catch (Exception e2) { } } } }
客戶端
package com.toov5.zkDubbo; import java.io.BufferedReader; import java.io.IOException; import java.io.InputStreamReader; import java.io.PrintWriter; import java.net.Socket; import java.util.ArrayList; import java.util.List; import org.I0Itec.zkclient.IZkChildListener; import org.I0Itec.zkclient.ZkClient; public class ZkServerClient { public static List<String> listServer = new ArrayList<String>(); public static void main(String[] args) { initServer(); ZkServerClient client= new ZkServerClient(); BufferedReader console = new BufferedReader(new InputStreamReader(System.in)); while (true) { String name; try { name = console.readLine(); if ("exit".equals(name)) { System.exit(0); } client.send(name); } catch (IOException e) { e.printStackTrace(); } } } // 註冊所有server public static void initServer() { listServer.clear(); // listServer.add("127.0.0.1:18080"); //連線服務 放到list中 存放叢集地址 做負載均衡用的 //建立zk連線 final ZkClient zkClient = new ZkClient("192.168.91.5",5000,10000); //讀取父節點資訊 final String root = "/toov5"; List<String> children=zkClient.getChildren(root); for(String pString : children){ //父+子 =完整的路徑 String path = root+"/"+pString; String nodeValue =zkClient.readData(path); //把值放到集合中去 listServer.add(nodeValue); } //服務發現成功 System.out.println("服務發現:"+listServer.toString()); //然後給 serverCount賦值 serverCount=listServer.size(); //使用zk事件監聽,如果服務發生宕機情況,重新讀取新的節點 zkClient.subscribeChildChanges(root, new IZkChildListener() { public void handleChildChange(String arg0, List<String> children) throws Exception { //children移除之後 剩下的 System.out.println("有服務宕機了,重新讀取新的節點資訊"+listServer.toString()); listServer.clear(); //清除之前的 然後讀取最新的 for(String pString : children){ //父+子 =完整的路徑 String path = root+"/"+pString; String nodeValue =zkClient.readData(path); //把值放到集合中去 listServer.add(nodeValue); } serverCount=children.size(); System.out.println("服務發現"+listServer.toString()); } }); } //請求總數 private static int reqCount = 1; //服務個數 private static int serverCount = 0; //初始值是0 // 獲取當前server資訊 public static String getServer() { // return listServer.get(0); 不能寫死 本地負載均衡輪詢演算法 String serverName = listServer.get(reqCount%serverCount ); System.out.println("客戶端請求次數:"+reqCount+"對應伺服器"+serverName); reqCount++; return serverName; } public void send(String name) { String server = ZkServerClient.getServer(); String[] cfg = server.split(":"); Socket socket = null; BufferedReader in = null; PrintWriter out = null; try { socket = new Socket(cfg[0], Integer.parseInt(cfg[1])); in = new BufferedReader(new InputStreamReader(socket.getInputStream())); out = new PrintWriter(socket.getOutputStream(), true); out.println(name); while (true) { String resp = in.readLine(); if (resp == null) break; else if (resp.length() > 0) { System.out.println("Receive : " + resp); break; } } } catch (Exception e) { e.printStackTrace(); } finally { if (out != null) { out.close(); } if (in != null) { try { in.close(); } catch (IOException e) { e.printStackTrace(); } } if (socket != null) { try { socket.close(); } catch (IOException e) { e.printStackTrace(); } } } } }
pom:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>zkDobbo</groupId> <artifactId>com.toov5.zkDubbo</artifactId> <version>0.0.1-SNAPSHOT</version> <dependencies> <dependency> <groupId>com.101tec</groupId> <artifactId>zkclient</artifactId> <version>0.8</version> </dependency> </dependencies> </project>
啟動三個服務端,一個客戶端。然後隨機關閉一個服務端:
控制檯:
當有新節點啟動時候:
核心部分程式碼欣賞~~