GitHub: github.com/jayknoxqu/r…

RMI簡介

? Java RMI，即 遠程方法調用(Remote Method Invocation)，一種用于實現遠程過程調用(RPC)(Remote procedure call)的Java API， 能直接傳輸序列化后的Java對象和分布式垃圾收集。它的實現依賴于Java虛擬機(JVM)，因此它僅支持從一個JVM到另一個JVM的調用。

rmi的實現

(1) 直接使用Registry實現rmi

服務端:

public class RegistryService {public static void main(String[] args) {try {// 本地主機上的遠程對象注冊表Registry的實例,默認端口1099Registry registry = LocateRegistry.createRegistry(1099);// 創建一個遠程對象HelloRegistryFacade hello = new HelloRegistryFacadeImpl();// 把遠程對象注冊到RMI注冊服務器上，并命名為HelloRegistryregistry.rebind("HelloRegistry", hello);System.out.println("======= 啟動RMI服務成功! =======");} catch (RemoteException e) {e.printStackTrace();}} } 復制代碼

接口:

繼承Remote接口

public interface HelloRegistryFacade extends Remote {String helloWorld(String name) throws RemoteException;} 復制代碼

接口實現:

繼承UnicastRemoteObject

public class HelloRegistryFacadeImpl extends UnicastRemoteObject implements HelloRegistryFacade{public HelloRegistryFacadeImpl() throws RemoteException {super();}@Overridepublic String helloWorld(String name) {return "[Registry] 你好! " + name;}} 復制代碼

客戶端:

public class RegistryClient {public static void main(String[] args) {try {Registry registry = LocateRegistry.getRegistry(1099);HelloRegistryFacade hello = (HelloRegistryFacade) registry.lookup("HelloRegistry");String response = hello.helloWorld("ZhenJin");System.out.println("=======> " + response + " <=======");} catch (NotBoundException | RemoteException e) {e.printStackTrace();}} } 復制代碼

圖解:

出處:https://www.tutorialspoint.com/java_rmi/java_rmi_introduction.htm

Registry(注冊表)是放置所有服務器對象的命名空間。 每次服務端創建一個對象時，它都會使用bind()或rebind()方法注冊該對象。 這些是使用稱為綁定名稱的唯一名稱注冊的。要調用遠程對象，客戶端需要該對象的引用,如(HelloRegistryFacade)。 即通過服務端綁定的名稱(HelloRegistry)從注冊表中獲取對象(lookup()方法)。 復制代碼

(2) 使用Naming方法實現rmi

服務端:

public class NamingService {public static void main(String[] args) {try {// 本地主機上的遠程對象注冊表Registry的實例LocateRegistry.createRegistry(1100);// 創建一個遠程對象HelloNamingFacade hello = new HelloNamingFacadeImpl();// 把遠程對象注冊到RMI注冊服務器上，并命名為Hello //綁定的URL標準格式為：rmi://host:port/nameNaming.bind("rmi://localhost:1100/HelloNaming", hello);System.out.println("======= 啟動RMI服務成功! =======");} catch (RemoteException | MalformedURLException | AlreadyBoundException e) {e.printStackTrace();}} } 復制代碼

接口和接口實現和Registry的方式一樣

客戶端:

public class NamingClient {public static void main(String[] args) {try {String remoteAddr="rmi://localhost:1100/HelloNaming";HelloNamingFacade hello = (HelloNamingFacade) Naming.lookup(remoteAddr);String response = hello.helloWorld("ZhenJin");System.out.println("=======> " + response + " <=======");} catch (NotBoundException | RemoteException | MalformedURLException e) {e.printStackTrace();}} } 復制代碼

Naming部分源碼:

public static Remote lookup(String name)throws NotBoundException,java.net.MalformedURLException,RemoteException{ParsedNamingURL parsed = parseURL(name);Registry registry = getRegistry(parsed);if (parsed.name == null)return registry;return registry.lookup(parsed.name); } 復制代碼

Naming其實是對Registry的一個封裝

Scala實現rmi

上面說了rmi是通過JVM虛擬機進行一個遠程調用的,我們通過Scala,kotlin等jvm語言印證下

服務端:

object ScalaRmiService extends App {try {val user:UserScalaFacade = new UserScalaFacadeImplLocateRegistry.createRegistry(1103)Naming.rebind("rmi://localhost:1103/UserScala", user)println("======= 啟動RMI服務成功! =======")} catch {case e: IOException => println(e)} } 復制代碼

接口

trait UserScalaFacade extends Remote {/*** 通過用戶名獲取用戶信息*/@throws(classOf[RemoteException])def getByName(userName: String): User/*** 通過用戶性別獲取用戶信息*/@throws(classOf[RemoteException])def getBySex(userSex: String): List[User]} 復制代碼

接口實現:

class UserScalaFacadeImpl extends UnicastRemoteObject with UserScalaFacade {/*** 模擬一個數據庫表*/private lazy val userList = List(new User("Jane", "女", 16),new User("jack", "男", 17),new User("ZhenJin", "男", 18))override def getByName(userName: String): User = userList.filter(u => userName.equals(u.userName)).headoverride def getBySex(userSex: String): List[User] = userList.filter(u => userSex.equals(u.userSex))} 復制代碼

實體類:

實體類必須實現序列化(Serializable)才能進行一個遠程傳輸

class User(name: String, sex: String, age: Int) extends Serializable {var userName: String = namevar userSex: String = sexvar userAge: Int = ageoverride def toString = s"User(userName=$userName, userSex=$userSex, userAge=$userAge)"} 復制代碼

Scala客戶端:

object ScalaRmiClient extends App {try {val remoteAddr="rmi://localhost:1103/UserScala"val userFacade = Naming.lookup(remoteAddr).asInstanceOf[UserScalaFacade]println(userFacade.getByName("ZhenJin"))System.out.println("--------------------------------------")for (user <- userFacade.getBySex("男")) println(user)} catch {case e: NotBoundException => println(e)case e: RemoteException => println(e)case e: MalformedURLException => println(e)}} 復制代碼

Java客戶端:

public class JavaRmiClient {public static void main(String[] args) {try {String remoteAddr="rmi://localhost:1103/UserScala";UserScalaFacade userFacade = (UserScalaFacade) Naming.lookup();User zhenJin = userFacade.getByName("ZhenJin");System.out.println(zhenJin);System.out.println("--------------------------------------");List<User> userList = userFacade.getBySex("男");System.out.println(userList);} catch (NotBoundException | RemoteException | MalformedURLException e) {e.printStackTrace();}} } 復制代碼

上面試驗可以證明Scala和Java是可以互通的,Scala本身也是可以直接引用Java類的

序列化簡介

? 序列化(Serialization)是將數據結構或對象狀態轉換為可以存儲（例如，在文件或存儲器緩沖區中）或傳輸（例如，通過網絡連接）的格式的過程, 反序列化(Deserialization)則是從一系列字節中提取數據結構的相反操作.

Kotlin實現rmi

服務端:

fun main(args: Array<String>) {try {val hello: HelloKotlinFacade = HelloKotlinFacadeImpl()LocateRegistry.createRegistry(1102)Naming.rebind("rmi://localhost:1101/HelloKotlin", hello)println("======= 啟動RMI服務成功! =======")} catch (e: IOException) {e.printStackTrace()} } 復制代碼

客戶端:

fun main(args: Array<String>) {try {val hello = Naming.lookup("rmi://localhost:1102/HelloKotlin") as HelloKotlinFacadeval response = hello.helloWorld("ZhenJin")println("=======> $response <=======")} catch (e: NotBoundException) {e.printStackTrace()} catch (e: RemoteException) {e.printStackTrace()} catch (e: MalformedURLException) {e.printStackTrace()} } 復制代碼

實現和接口省略...

SpringBoot實現rmi

StringBoot通過配置就可以簡單實現rmi了

服務端:

@Configuration public class RmiServiceConfig {@Beanpublic RmiServiceExporter registerService(UserFacade userFacade) {RmiServiceExporter rmiServiceExporter = new RmiServiceExporter();rmiServiceExporter.setServiceName("UserInfo");rmiServiceExporter.setService(userFacade);rmiServiceExporter.setServiceInterface(UserFacade.class);rmiServiceExporter.setRegistryPort(1101);return rmiServiceExporter;} } 復制代碼

客戶端:

@Configuration public class RmiClientConfig {@Beanpublic UserFacade userInfo() {RmiProxyFactoryBean rmiProxyFactoryBean = new RmiProxyFactoryBean();rmiProxyFactoryBean.setServiceUrl("rmi://localhost:1101/UserInfo");rmiProxyFactoryBean.setServiceInterface(UserFacade.class);rmiProxyFactoryBean.afterPropertiesSet();return (UserFacade) rmiProxyFactoryBean.getObject();}} 復制代碼

客戶端測試類:

@Autowired private UserFacade userFacade;@Test public void userBySexTest() {try {List<User> userList = userFacade.getBySex("男");userList.forEach(System.out::println);} catch (RemoteException e) {e.printStackTrace();} } 復制代碼

通過測試類可以看出,這和我們平時的程序調用內部方法沒什么區別!

rmi調用過程

大家可以通過下面文章加深了解:

stuff.mit.edu/afs/athena/…

• 有兩個遠程服務接口可供client調用，Factory和Product接口

• FactoryImpl類實現了Factory接口，ProductImpl類實現了Product接口

  1. FactoryImpl被注冊到了rmi-registry中 2. client端請求一個Factory的引用 3. rmi-registry返回client端一個FactoryImpl的引用 4. client端調用FactoryImpl的遠程方法請求一個ProductImpl的遠程引用 5. FactoryImpl返回給client端一個ProductImpl引用 6. client通過ProductImpl引用調用遠程方法 復制代碼

socket工廠文檔: docs.oracle.com/javase/8/do…

Zookeeper實現rmi

出處:http://www.importnew.com/20344.html

安裝Zookeeper

解壓 ZooKeeper

tar -zxvf zookeeper-3.4.12.tar.gz 復制代碼

在 conf 目錄新建 zoo.cfg

cd zookeeper-3.4.12/conf vim zoo.cfg 復制代碼

zoo.cfg 代碼如下（自己指定 log 文件目錄）：

tickTime=2000 dataDir=/usr/local/zookeeper-3.4.12/data dataLogDir=/usr/local/zookeeper-3.4.12/log clientPort=2181 復制代碼

在 bin 目錄下，啟動 Zookeeper：

cd zookeeper-3.4.12/bin ./zkServer.sh start 復制代碼

消費者:

public class RmiConsumer {// 用于等待 SyncConnected 事件觸發后繼續執行當前線程private CountDownLatch latch = new CountDownLatch(1);// 定義一個 volatile 成員變量，用于保存最新的 RMI 地址（考慮到該變量或許會被其它線程所修改，一旦修改后，該變量的值會影響到所有線程）private volatile List<String> urlList = new ArrayList<>();// 構造器public RmiConsumer() {ZooKeeper zk = connectServer(); // 連接 ZooKeeper 服務器并獲取 ZooKeeper 對象if (zk != null) {watchNode(zk); // 觀察 /registry 節點的所有子節點并更新 urlList 成員變量}}// 查找 RMI 服務public <T extends Remote> T lookup() {T service = null;int size = urlList.size();if (size > 0) {String url;if (size == 1) {url = urlList.get(0); // 若 urlList 中只有一個元素，則直接獲取該元素log.debug("using only url: {}", url);} else {url = urlList.get(ThreadLocalRandom.current().nextInt(size)); // 若 urlList 中存在多個元素，則隨機獲取一個元素log.debug("using random url: {}", url);}service = lookupService(url); // 從 JNDI 中查找 RMI 服務}return service;}// 連接 ZooKeeper 服務器private ZooKeeper connectServer() {ZooKeeper zk = null;try {zk = new ZooKeeper(Constant.ZK_CONNECTION_STRING, Constant.ZK_SESSION_TIMEOUT, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getState() == Event.KeeperState.SyncConnected) {latch.countDown(); // 喚醒當前正在執行的線程}}});latch.await(); // 使當前線程處于等待狀態} catch (IOException | InterruptedException e) {log.error("", e);}return zk;}// 觀察 /registry 節點下所有子節點是否有變化private void watchNode(final ZooKeeper zk) {try {List<String> nodeList = zk.getChildren(Constant.ZK_REGISTRY_PATH, event -> {if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {watchNode(zk); // 若子節點有變化，則重新調用該方法（為了獲取最新子節點中的數據）}});List<String> dataList = new ArrayList<>(); // 用于存放 /registry 所有子節點中的數據for (String node : nodeList) {byte[] data = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null); // 獲取 /registry 的子節點中的數據dataList.add(new String(data));}log.debug("node data: {}", dataList);urlList = dataList; // 更新最新的 RMI 地址} catch (KeeperException | InterruptedException e) {log.error("", e);}}// 在 JNDI 中查找 RMI 遠程服務對象@SuppressWarnings("unchecked")private <T> T lookupService(String url) {T remote = null;try {remote = (T) Naming.lookup(url);} catch (NotBoundException | MalformedURLException | RemoteException e) {log.error("遠程查找出錯!", e);}return remote;} } 復制代碼

生產者:

public class RmiProvider {/*** 用于等待 SyncConnected 事件觸發后繼續執行當前線程*/private CountDownLatch latch = new CountDownLatch(1);// 發布 RMI 服務并注冊 RMI 地址到 ZooKeeper 中public void publish(Remote remote, String host, int port) {String url = publishService(remote, host, port); // 發布 RMI 服務并返回 RMI 地址if (url != null) {ZooKeeper zk = connectServer(); // 連接 ZooKeeper 服務器并獲取 ZooKeeper 對象if (zk != null) {createNode(zk, url); // 創建 ZNode 并將 RMI 地址放入 ZNode 上}}}/***發布 RMI 服務*/private String publishService(Remote remote, String host, int port) {String url = null;try {url = String.format("rmi://%s:%d/%s", host, port, remote.getClass().getName());LocateRegistry.createRegistry(port);Naming.rebind(url, remote);log.debug("publish rmi service (url: {})", url);} catch (RemoteException | MalformedURLException e) {log.error("", e);}return url;}// 連接 ZooKeeper 服務器private ZooKeeper connectServer() {ZooKeeper zk = null;try {zk = new ZooKeeper(Constant.ZK_CONNECTION_STRING, Constant.ZK_SESSION_TIMEOUT, new Watcher() {@Overridepublic void process(WatchedEvent event) {if (event.getState() == Event.KeeperState.SyncConnected) {latch.countDown(); // 喚醒當前正在執行的線程}}});latch.await(); // 使當前線程處于等待狀態} catch (IOException | InterruptedException e) {log.error("", e);}return zk;}/*** 創建節點*/private void createNode(ZooKeeper zk, String url) {try {byte[] data = url.getBytes();String path = zk.create(Constant.ZK_PROVIDER_PATH, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 創建一個臨時性且有序的 ZNodelog.debug("create zookeeper node ({} => {})", path, url);} catch (KeeperException | InterruptedException e) {log.error("", e);}} } 復制代碼

圖解:

代碼已上傳到GitHub上:github.com/jayknoxqu/r…

轉載于:https://juejin.im/post/5c149fa25188250deb75b411

總結

以上是生活随笔為你收集整理的分布式架构基础:Java RMI详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。