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);
//zookeeper服务 private ZooKeeper zk;
//是否注册过端口 private boolean isRegistry = false; public RMIServer(){ zk = connectServer(); } // 同步停等 private CountDownLatch latch = new CountDownLatch(1); // 发布RMI服务 public void publish(Remote remote,String key) {
if(remote==null||key==null){
return;
} String url = publishService(remote); if (url != null&& zk!=null) { createNode(zk,url,key); } } // 发布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 (Exception 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 (Exception e) { logger.error("", e); } if(zk!=null){//建立znode节点 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 (Exception e) { e.printStackTrace(); } } return zk; } //创建 ZNode private void createNode(ZooKeeper zk, String url,String key) { try { byte[] data = (key+"#:#"+url).getBytes();//一个系统可能有多个不同对外方法,通过key值进行映射 //创建一个临时有序的数据节点 String path = zk.create(ConfigHelp.ZK_RMI_PATH+"/", data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); logger.info("create zookeeper node:"+path+"========="+url); } catch (Exception 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(Integer.MAX_VALUE); }}
3.RMIClient
public class RMIClient { private static final Logger logger = LoggerFactory.getLogger(RMIClient.class); // private CountDownLatch latch = new CountDownLatch(1); // key值对应的URL连接集合 private volatile HashMap<String,List<String>> dataMap = new HashMap<String, List<String>>(); public RMIClient() { ZooKeeper zk = connectServer(); // 连接ZooKeeper watchNode(zk); } // 查找 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 (Exception e) { logger.error("", e); } return zk; } // 观察 /registry 节点下所有子节点是否有变化 private void watchNode(final ZooKeeper zk) { if(zk == null){ return; } 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(zk); // 若子节点有变化,则重新调用该方法(为了获取最新子节点中的数据) } } }); List<String> dataList = new ArrayList<>(); // 用于存放 /root所有子节点中的数据 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); // 获取 /root/rmi 的子节点中的数据 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); } }this.dataMap = dataMap; } catch (Exception e) { logger.error("", e); } } // 在 JNDI 中查找 RMI 远程服务对象private <T> T lookupService(String url,String key) { T remote = null; try { remote = (T) Naming.lookup(url); } catch (Exception e) { if (e instanceof ConnectException) { 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(); } } }}
原标题:基于zookeeper的远程方法调用(RMI)的实现
关键词: