[TOC] # 服務器上下線動態感知 ## 需求描述 某分布式系統中，主節點可以有多臺，可以動態上下線 任意一臺客戶端都能實時感知到主節點服務器的上下線 ## 設計思路   ## 代碼開發 ### 服務端實現 ~~~ package hello; import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat; import java.io.IOException; import java.util.concurrent.CountDownLatch; public class DistributedServer { private static final String CONNECT_STRING = "192.168.33.12:2181,192.168.3.33:2181"; //超過2秒就認為掛了 private static final int SESSION_TIME_OUT = 2000; //父節點 private static final String PARENT_NODE = "/servers"; private ZooKeeper zk = null; //latch就相當于一個對象,當latch.await()方法執行時,線程就會等待 //當latch的count減為0的時候,將會喚醒等待的線程 //讓主線程阻塞掉 private CountDownLatch countDownLatch = new CountDownLatch(1); //創建到zk客戶端的連接 private void getConnect() throws IOException, InterruptedException { //一new完就往下走,但是這時候客戶端還沒完成連接,所以我們要等他創建好 //一旦成功握手這邊的process就會回調一次 zk = new ZooKeeper(CONNECT_STRING, SESSION_TIME_OUT, new Watcher() { public void process(WatchedEvent watchedEvent) { //SyncConnected同步連接 //回調了并且事件等于連接成功 if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { //把計數減少,然后主線程就可以往下走了 countDownLatch.countDown(); } //收到事件通知后的回調函數(應該是我們自己的事件處理邏輯) System.out.println(watchedEvent.getType() + "---" + watchedEvent.getPath()); try { //再次注冊監聽,監聽/下面的 zk.getChildren("/", true); } catch (Exception e) { e.printStackTrace(); } } }); countDownLatch.await(); } //向zk集群注冊服務信息 private void registerServer(String hostname) throws KeeperException, InterruptedException { //判斷這個父節點是否存在 Stat exists = zk.exists(PARENT_NODE, false); /* * Ids.OPEN_ACL_UNSAFE 創建開放節點，允許任意操作 * Ids.READ_ACL_UNSAFE 創建只讀節點 * Ids.CREATOR_ALL_AC /創建者有全部權限 * */ if (exists == null) { //如果這個父節點/servers不存在就把他創建出來,父節點需要持久的,其他注冊過來的就用短暫的 zk.create(PARENT_NODE, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } //創建子節點,并且這個子節點后面是跟序號的,在子節點的名稱后面加上一串數字后綴,避免沖突,而且子節點是短暫的 String path = zk.create(PARENT_NODE + "/server", hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //打印下表示這臺服務器上線了 System.out.println(hostname + "is online..." + path); } //業務功能 public void handleBussiness(String hostname) throws InterruptedException { //這邊業務就是打印一句話,那個ip開始工作了 System.out.println(hostname + "start working ..."); //讓主線程sleep Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws IOException, InterruptedException, KeeperException { DistributedServer server = new DistributedServer(); //獲取zk連接 server.getConnect(); //利用zk連接注冊服務器信息(主機名),這邊參數可能用args[0]來替代 server.registerServer("192.168.3.33"); //啟動業務功能 server.handleBussiness("192.168.3.33"); } } ~~~ ### 客戶端實現 ~~~ package hello; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; public class DistributedClient { private static final String connectString = "192.168.33.12:2181,192.168.3.33:2181"; //超過2秒就認為掛了 private static final int sessionTimeOut = 2000; //父節點 private static final String parentNode = "/servers"; //提供給各業務線程使用的服務器列表 private volatile List<String> serverList; private ZooKeeper zk = null; //latch就相當于一個對象,當latch.await()方法執行時,線程就會等待 //當latch的count減為0的時候,將會喚醒等待的線程 //讓主線程阻塞掉 private CountDownLatch countDownLatch = new CountDownLatch(1); //創建到zk的客戶端連接 public void getConnect() throws IOException, InterruptedException { //一new完就往下走,但是這時候客戶端還沒完成連接,所以我們要等他創建好 //一旦成功握手這邊的process就會回調一次 zk = new ZooKeeper(connectString, sessionTimeOut, new Watcher() { public void process(WatchedEvent watchedEvent) { //SyncConnected同步連接 //回調了并且事件等于連接成功 if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { //把計數減少,然后主線程就可以往下走了 countDownLatch.countDown(); } //收到事件通知后的回調函數(應該是我們自己的事件處理邏輯) System.out.println(watchedEvent.getType() + "---" + watchedEvent.getPath()); try { //重新更新服務器列表，并且注冊了監聽 getServerList(); } catch (Exception e) { e.printStackTrace(); } } }); countDownLatch.await(); } //獲取服務器信息列表 private void getServerList() throws Exception { //獲取服務器子節點信息,并且對父節點進行監聽 List<String> children = zk.getChildren(parentNode, true); //先創建一個局部的list來存服務器信息 List<String> servers = new ArrayList<String>(); for (String child : children) { //child只是子節點的節點名,我們獲取數據要給全路徑 //結果是服務器的名字 byte[] data = zk.getData(parentNode + "/" + child, false, null); servers.add(new String(data)); } //把servers賦值給成員變量serverList,已提供給各業務線程使用 //避免別的線程正在看的時候,這邊正在寫,所以弄個賦值 serverList = servers; //打印服務器列表 System.out.println(serverList); } //業務線程 public void handleBussiness() throws InterruptedException { //這邊業務就是打印一句話,那個ip開始工作了 System.out.println("client start working ..."); //讓主線程sleep Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { //獲取zk連接 DistributedClient client = new DistributedClient(); client.getConnect(); //獲取servers的子節點信息(并監聽),從中獲取服務器信息列表 client.getServerList(); //業務線程啟動 client.handleBussiness(); } } ~~~ 然后可以打成jar包放到別的電腦上