基於zookeeper的遠程方法調用(RMI)的實現

採用zookeeper的命令服務，採用不一樣的目錄結構存儲不一樣模塊不一樣服務的rmi的url，使用key來對應不一樣的服務。同時採用zookeeper解決了單點問題。

當有兩個相同的服務註冊時，由於採用的是臨時有序znode，也會註冊成功，客戶端在調用時，發現有多個不一樣url，則會隨機調取一個使用。

當一個服務down時，zookeeper會通知客戶端更新url緩存，從而去掉down服務的連接。加入時，也會同步客戶端url數據。

原理圖以下：

1).目錄結構

2).zookeeper調用

3).分佈式調用

實現代碼以下：

1.定義測試服務接口與實現

public interface Service extends Remote,Serializable{ public void service(String name); } 

public class ServiceImpl implements Service{ @Override public void service(String name) { System.out.println(name); } }

2.RMIServer

public class RMIServer { private static final Logger logger = LoggerFactory .getLogger(RMIServer.class); private static ZooKeeper zk; private boolean isRegistry = false; //同步鎖 private Lock _lock = new ReentrantLock(); private static final Map<String,String> CACHED_URL = new HashMap<String,String>(); public RMIServer() { zk = connectServer(); new Thread(new Runnable() { @Override public void run() { while (true) { try { Thread.currentThread().sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } logger.info("check zk..."); _lock.lock(); if (zk != null) { if (zk.getState().isAlive() && zk.getState().isConnected()) { logger.info("zk is ok"); _lock.unlock(); continue; } } close(); logger.info("reConnectServer ..."); zk = connectServer(); Iterator<String> it = CACHED_URL.keySet().iterator(); while(it.hasNext()){ String key = it.next(); String url = CACHED_URL.get(key); createNode(zk,url,key); } logger.info("reConnectServer ok"); _lock.unlock(); } } private void close() { if(zk!=null){ try { zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } zk = null; } } }).start(); } // 用於等待 SyncConnected 事件觸發後繼續執行當前線程 private CountDownLatch latch = new CountDownLatch(1); // 發佈 RMI 服務並註冊 RMI 地址到 ZooKeeper 中 public void publish(Remote remote, String key) { _lock.lock(); String url = publishService(remote); // 發佈 RMI 服務並返回 RMI 地址 if (url != null) { if (zk != null) { if (zk.getState().isAlive() && zk.getState().isConnected()) { } else { if(zk!=null){ try { zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } zk = null; } zk = connectServer(); } createNode(zk, url, key); // 建立 ZNode 並將 RMI 地址放入 ZNode 上  } } _lock.unlock(); } // 發佈 RMI 服務 private String publishService(Remote remote) { String url = null; try { String host = ConfigHelp.getLocalConifg("rmiIP", "127.0.0.1"); int port = Integer.valueOf(ConfigHelp.getLocalConifg("rmiPort", "10990")); url = String.format("rmi://%s:%d/%s", host, port, remote.getClass() .getName()); if (!isRegistry) { LocateRegistry.createRegistry(port); isRegistry = true; } Naming.rebind(url, remote); logger.debug("publish rmi service (url: {})", url); } catch (RemoteException | MalformedURLException e) { logger.error("", e); } return url; } // 鏈接 ZooKeeper 服務器 private ZooKeeper connectServer() { ZooKeeper zk = null; try { zk = new ZooKeeper(ConfigHelp.ZK_CONNECTION_STRING, ConfigHelp.ZK_SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getState() == Event.KeeperState.SyncConnected) { latch.countDown(); // 喚醒當前正在執行的線程  } } }); latch.await(); // 使當前線程處於等待狀態 } catch (IOException | InterruptedException e) { logger.error("", e); } if (zk != null) { try { Stat stat = zk.exists(ConfigHelp.ZK_ROOT_PATH, false); if (stat == null) { String path = zk.create(ConfigHelp.ZK_ROOT_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); // 建立一個臨時性且有序的 ZNode logger.info("create zookeeper node ({})", path); } stat = zk.exists(ConfigHelp.ZK_RMI_PATH, false); if (stat == null) { String path = zk.create(ConfigHelp.ZK_RMI_PATH, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); // 建立一個臨時性且有序的 ZNode logger.info("create zookeeper node ({})", path); } } catch (KeeperException | InterruptedException e) { e.printStackTrace(); } } return zk; } // 建立 ZNode private void createNode(ZooKeeper zk, String url, String key) { try { CACHED_URL.put(key, url); byte[] data = (key + "#:#" + url).getBytes(); String path = zk.create(ConfigHelp.ZK_RMI_PATH + "/", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 建立一個臨時性且有序的 ZNode logger.info("create zookeeper node ({} => {})", path, url); } catch (KeeperException | InterruptedException e) { logger.error("", e); e.printStackTrace(); } } public static void main(String[] args) throws Exception { RMIServer server = new RMIServer(); Service service = new ServiceImpl(); server.publish(service, "Test"); server.publish(service, "Test1"); Thread.currentThread().sleep(50000); server.publish(service, "Test3"); Thread.currentThread().sleep(Integer.MAX_VALUE); } }

3.RMIClient

public class RMIClient { private static final Logger logger = LoggerFactory.getLogger(RMIClient.class); // 用於等待 SyncConnected 事件觸發後繼續執行當前線程 private CountDownLatch latch = new CountDownLatch(1); // 定義一個 volatile 成員變量，用於保存最新的 RMI 地址（考慮到該變量或許會被其它線程所修改，一旦修改後，該變量的值會影響到全部線程） private volatile HashMap<String,List<String>> dataMap = new HashMap<String, List<String>>(); private Lock _lock = new ReentrantLock(); private static ZooKeeper zk; // 構造器 public RMIClient() { zk = connectServer(); // 鏈接 ZooKeeper 服務器並獲取 ZooKeeper 對象  watchNode(); new Thread(new Runnable() { @Override public void run() { while (true) { try { Thread.currentThread().sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } _lock.lock(); if (zk != null) { if (zk.getState().isAlive() && zk.getState().isConnected()) { _lock.unlock(); continue; } } if(zk!=null){ try { zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } zk = null; } zk = connectServer(); _lock.unlock(); } } }).start(); } // 查找 RMI 服務 public <T extends Remote> T lookup(String key) { T service = null; int size = dataMap.size(); if (size > 0) { String url = null; if(dataMap.containsKey(key)){ List<String> urlList = dataMap.get(key); if(urlList.size()>0){ if(urlList.size()==1){ url = urlList.get(0); }else{ url = urlList.get(ThreadLocalRandom.current().nextInt(size)); } } service = lookupService(url,key); // 從 JNDI 中查找 RMI 服務  } } return service; } // 鏈接 ZooKeeper 服務器 private ZooKeeper connectServer() { ZooKeeper zk = null; try { zk = new ZooKeeper(ConfigHelp.ZK_CONNECTION_STRING, ConfigHelp.ZK_SESSION_TIMEOUT, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getState() == Event.KeeperState.SyncConnected) { latch.countDown(); // 喚醒當前正在執行的線程  } } }); latch.await(); // 使當前線程處於等待狀態 } catch (IOException | InterruptedException e) { logger.error("", e); } return zk; } // 觀察 /registry 節點下全部子節點是否有變化 private void watchNode() { _lock.lock(); if(zk!=null&&zk.getState().isAlive()&&zk.getState().isConnected()){ }else{ if(zk!=null){ try { zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } zk = null; } zk = connectServer(); } try { List<String> nodeList = zk.getChildren(ConfigHelp.ZK_RMI_PATH, new Watcher() { @Override public void process(WatchedEvent event) { if (event.getType() == Event.EventType.NodeChildrenChanged) { watchNode(); // 若子節點有變化，則從新調用該方法（爲了獲取最新子節點中的數據）  } } }); List<String> dataList = new ArrayList<>(); // 用於存放 /registry 全部子節點中的數據 HashMap<String,List<String>> dataMap = new HashMap<String, List<String>>(); for (String node : nodeList) { byte[] data = zk.getData(ConfigHelp.ZK_RMI_PATH + "/" + node, false, null); // 獲取 /registry 的子節點中的數據 dataList.add(new String(data)); String d = new String(data).toString(); String key = d.split("#:#")[0]; String url = d.split("#:#")[1]; if(dataMap.containsKey(key)){ dataMap.get(key).add(url); }else{ List<String> list = new ArrayList<String>(); list.add(url); dataMap.put(key, list); } } logger.debug("node data: {}", dataList); this.dataMap = dataMap; } catch (KeeperException | InterruptedException e) { logger.error("", e); } _lock.unlock(); } // 在 JNDI 中查找 RMI 遠程服務對象 @SuppressWarnings("unchecked") private <T> T lookupService(String url,String key) { T remote = null; try { remote = (T) Naming.lookup(url); } catch (NotBoundException | MalformedURLException | RemoteException e) { if (e instanceof ConnectException) { // 若鏈接中斷，則使用 urlList 中第一個 RMI 地址來查找（這是一種簡單的重試方式，確保不會拋出異常） logger.error("ConnectException -> url: {}", url); if(dataMap.containsKey(key)){ List<String> urlList = dataMap.get(key); if(urlList.size()>0){ return lookupService(urlList.get(0),key); } } } logger.error("", e); } return remote; } public static void main(String[] args) { RMIClient client = new RMIClient(); while(true){ Service service = client.lookup("Test"); service.service("test12"); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } } } }

4.RMIHelper

public class RMIHelper { private static final RMIServer SERVER = new RMIServer(); private static final RMIClient CLIENT = new RMIClient(); public static synchronized void publish(Remote remote,String key){ SERVER.publish(remote, key); } public static synchronized <T extends Remote> T lookup(String key){ return CLIENT.lookup(key); } public static void main(String[] args) throws Exception { while(true){ Service service = RMIHelper.lookup("Test"); service.service("test12"); Service service1 = RMIHelper.lookup("Test1"); service1.service("test12"); try { Thread.sleep(2000); } catch (Exception e) { e.printStackTrace(); } } } }