你的位置:首页 > Java教程

[Java教程]基于zookeeper的远程方法调用(RMI)的实现


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();      }    }      }}