以前在zookeeper介紹時,有提到過zookeeper的應用場景有服務器上下線動態感知、分佈式共享鎖等等。本節博主將爲你們提供一個分佈式服務動態感知的事例,在後期博主的開源項目中的本身開發遠程分佈式框架rpc服務也將使用。因此,你們必定要掌握這塊知識點。java
對於還不清楚動態感知原理的請先看以前的文章大數據教程(3.3):zookeeper簡介。apache
(1)分佈式服務端註冊centos
package com.empire.zookeeper.test.em_zookeeper.zkdist; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.data.Stat; import org.apache.zookeeper.ZooKeeper; public class DistributedServer { private static final String connectString = "centos-aaron-07:2181,centos-aaron-08:2181,centos-aaron-16:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; private CountDownLatch connectedSemaphore = new CountDownLatch(1); private ZooKeeper zk = null; /** * 建立到zk的客戶端鏈接 * * @throws Exception */ public void getConnect() throws Exception { zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知後的回調函數(應該是咱們本身的事件處理邏輯) System.out.println(event.getType() + "---" + event.getPath()); try { zk.getChildren("/", true); connectedSemaphore.countDown(); } catch (Exception e) { } } }); connectedSemaphore.await(); } /** * 向zk集羣註冊服務器信息 * * @param hostname * @throws Exception */ public void registerServer(String hostname) throws Exception { Stat stat = zk.exists(parentNode, false); System.out.println(stat==null?"not exist":"exist"); if(stat==null) { String parentNodeCreate = zk.create(parentNode , null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println(hostname + "is online.." + create); } /** * 業務功能 * * @throws InterruptedException */ public void handleBussiness(String hostname) throws InterruptedException { System.out.println(hostname + "start working....."); Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { // 獲取zk鏈接 DistributedServer server = new DistributedServer(); server.getConnect(); // 利用zk鏈接註冊服務器信息 server.registerServer(args[0]); // 啓動業務功能 server.handleBussiness(args[0]); } }
(2)分佈式客戶端感知服務狀態服務器
package com.empire.zookeeper.test.em_zookeeper.zkdist; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper; public class DistributedClient { private static final String connectString = "centos-aaron-07:2181,centos-aaron-08:2181,centos-aaron-16:2181"; private static final int sessionTimeout = 2000; private static final String parentNode = "/servers"; // 注意:加volatile的意義何在? private volatile List<String> serverList; private CountDownLatch connectedSemaphore = new CountDownLatch(1); private ZooKeeper zk = null; /** * 建立到zk的客戶端鏈接 * * @throws Exception */ public void getConnect() throws Exception { zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() { @Override public void process(WatchedEvent event) { // 收到事件通知後的回調函數(應該是咱們本身的事件處理邏輯) try { //從新更新服務器列表,而且註冊了監聽 getServerList(); connectedSemaphore.countDown(); } catch (Exception e) { } } }); connectedSemaphore.await(); } /** * 獲取服務器信息列表 * * @throws Exception */ public 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); } /** * 業務功能 * * @throws InterruptedException */ public void handleBussiness() throws InterruptedException { System.out.println("client start working....."); 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(); } }
注意點:在啓動server端上線時,須要輸入啓動參數hostname來啓動,該值用來惟一標識當前上線的是哪臺服務。session
最後總結:其實分佈式服務動態感知主要運行zookeeper的短暫節點的的特性(服務器下線,zookeeper會自動刪除該服務器對於的臨時節點信息,並通知其它監聽了該節點變化的客戶端)來實現感知動態下線,以及zookeeper的節點監聽事件功能來實現感知服務動態上線。框架
最後寄語,以上是博主本次文章的所有內容,若是你們以爲博主的文章還不錯,請點贊;若是您對博主其它服務器技術或者博主本人感興趣,請關注博主博客,而且歡迎隨時跟博主溝通交流。分佈式