布式架構基礎:Java RMI詳解
RMI簡介 Java RMI，即 遠端方法呼叫(Remote Method Invocation)，一種用於實現遠端過程呼叫(RPC)(Remote procedure call)的Java API， 能直接傳輸序列化後的Java物件和分散式垃圾收集。它的實現依賴於Java虛擬機器(JVM)，因此它僅支援從一個JVM到另一個JVM的呼叫。
public class RegistryService { public static void main(String[] args) { try { // 本地主機上的遠端物件登錄檔Registry的例項,預設埠1099 Registry registry = LocateRegistry.createRegistry(1099); // 建立一個遠端物件 HelloRegistryFacade hello = new HelloRegistryFacadeImpl(); // 把遠端物件註冊到RMI註冊伺服器上，並命名為HelloRegistry registry.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(); } @Override public 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(); } } }
圖解:
要呼叫遠端物件，客戶端需要該物件的引用,如(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/name Naming.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 UserScalaFacadeImpl LocateRegistry.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)).head
override 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 = name var userSex: String = sex var userAge: Int = age override def toString = s”User(userName=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)則是從一系列位元組中提取資料結構的相反操作.
fun main(args: Array) { 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) { try { val hello = Naming.lookup(“rmi://localhost:1102/HelloKotlin”) as HelloKotlinFacade val 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 { @Bean public 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 {
@Bean public 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 userList = userFacade.getBySex(“男”); userList.forEach(System.out::println); } catch (RemoteException e) { e.printStackTrace(); } } 通過測試類可以看出,這和我們平時的程式呼叫內部方法沒什麼區別!
rmi呼叫過程
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引用呼叫遠端方法 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() { @Override public 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() {
@Override
public 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); // 建立一個臨時性且有序的 ZNode
log.debug("create zookeeper node ({} => {})", path, url);
} catch (KeeperException | InterruptedException e) {
log.error("", e);
}
}
}
圖解:
