采用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("rmisession_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();            }        }            }}