GitHub: https://github.com/jayknoxqu/rmi-example
RMI簡(jiǎn)介
? Java RMI,即 遠(yuǎn)程方法調(diào)用(Remote Method Invocation)涨薪,一種用于實(shí)現(xiàn)遠(yuǎn)程過程調(diào)用(RPC)(Remote procedure call)的Java API骑素, 能直接傳輸序列化后的Java對(duì)象和分布式垃圾收集。它的實(shí)現(xiàn)依賴于Java虛擬機(jī)(JVM)刚夺,因此它僅支持從一個(gè)JVM到另一個(gè)JVM的調(diào)用献丑。
rmi的實(shí)現(xiàn)
(1) 直接使用Registry實(shí)現(xiàn)rmi
服務(wù)端:
public class RegistryService {
public static void main(String[] args) {
try {
// 本地主機(jī)上的遠(yuǎn)程對(duì)象注冊(cè)表Registry的實(shí)例,默認(rèn)端口1099
Registry registry = LocateRegistry.createRegistry(1099);
// 創(chuàng)建一個(gè)遠(yuǎn)程對(duì)象
HelloRegistryFacade hello = new HelloRegistryFacadeImpl();
// 把遠(yuǎn)程對(duì)象注冊(cè)到RMI注冊(cè)服務(wù)器上,并命名為HelloRegistry
registry.rebind("HelloRegistry", hello);
System.out.println("======= 啟動(dòng)RMI服務(wù)成功! =======");
} catch (RemoteException e) {
e.printStackTrace();
}
}
}
接口:
繼承Remote接口
public interface HelloRegistryFacade extends Remote {
String helloWorld(String name) throws RemoteException;
}
接口實(shí)現(xiàn):
繼承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();
}
}
}
圖解:
出處:https://www.tutorialspoint.com/java_rmi/java_rmi_introduction.htm
Registry(注冊(cè)表)是放置所有服務(wù)器對(duì)象的命名空間侠姑。
每次服務(wù)端創(chuàng)建一個(gè)對(duì)象時(shí)创橄,它都會(huì)使用bind()或rebind()方法注冊(cè)該對(duì)象。
這些是使用稱為綁定名稱的唯一名稱注冊(cè)的结借。
要調(diào)用遠(yuǎn)程對(duì)象筐摘,客戶端需要該對(duì)象的引用,如(HelloRegistryFacade)。
即通過服務(wù)端綁定的名稱(HelloRegistry)從注冊(cè)表中獲取對(duì)象(lookup()方法)船老。
(2) 使用Naming方法實(shí)現(xiàn)rmi
服務(wù)端:
public class NamingService {
public static void main(String[] args) {
try {
// 本地主機(jī)上的遠(yuǎn)程對(duì)象注冊(cè)表Registry的實(shí)例
LocateRegistry.createRegistry(1100);
// 創(chuàng)建一個(gè)遠(yuǎn)程對(duì)象
HelloNamingFacade hello = new HelloNamingFacadeImpl();
// 把遠(yuǎn)程對(duì)象注冊(cè)到RMI注冊(cè)服務(wù)器上咖熟,并命名為Hello
//綁定的URL標(biāo)準(zhǔn)格式為:rmi://host:port/name
Naming.bind("rmi://localhost:1100/HelloNaming", hello);
System.out.println("======= 啟動(dòng)RMI服務(wù)成功! =======");
} catch (RemoteException | MalformedURLException | AlreadyBoundException e) {
e.printStackTrace();
}
}
}
接口和接口實(shí)現(xiàn)和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其實(shí)是對(duì)Registry的一個(gè)封裝
Scala實(shí)現(xiàn)rmi
上面說了rmi是通過JVM虛擬機(jī)進(jìn)行一個(gè)遠(yuǎn)程調(diào)用的,我們通過Scala,kotlin等jvm語言印證下
服務(wù)端:
object ScalaRmiService extends App {
try {
val user:UserScalaFacade = new UserScalaFacadeImpl
LocateRegistry.createRegistry(1103)
Naming.rebind("rmi://localhost:1103/UserScala", user)
println("======= 啟動(dòng)RMI服務(wù)成功! =======")
} 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]
}
接口實(shí)現(xiàn):
class UserScalaFacadeImpl extends UnicastRemoteObject with UserScalaFacade {
/**
* 模擬一個(gè)數(shù)據(jù)庫表
*/
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))
}
實(shí)體類:
實(shí)體類必須實(shí)現(xiàn)序列化(Serializable)才能進(jìn)行一個(gè)遠(yuǎn)程傳輸
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=$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();
}
}
}
上面試驗(yàn)可以證明Scala和Java是可以互通的,Scala本身也是可以直接引用Java類的
序列化簡(jiǎn)介
? 序列化(Serialization)是將數(shù)據(jù)結(jié)構(gòu)或?qū)ο鬆顟B(tài)轉(zhuǎn)換為可以存儲(chǔ)(例如,在文件或存儲(chǔ)器緩沖區(qū)中)或傳輸(例如柳畔,通過網(wǎng)絡(luò)連接)的格式的過程, 反序列化(Deserialization)則是從一系列字節(jié)中提取數(shù)據(jù)結(jié)構(gòu)的相反操作.
Kotlin實(shí)現(xiàn)rmi
服務(wù)端:
fun main(args: Array<String>) {
try {
val hello: HelloKotlinFacade = HelloKotlinFacadeImpl()
LocateRegistry.createRegistry(1102)
Naming.rebind("rmi://localhost:1101/HelloKotlin", hello)
println("======= 啟動(dòng)RMI服務(wù)成功! =======")
} catch (e: IOException) {
e.printStackTrace()
}
}
客戶端:
fun main(args: Array<String>) {
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()
}
}
實(shí)現(xiàn)和接口省略...
SpringBoot實(shí)現(xiàn)rmi
StringBoot通過配置就可以簡(jiǎn)單實(shí)現(xiàn)rmi了
服務(wù)端:
@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();
}
}
客戶端測(cè)試類:
@Autowired
private UserFacade userFacade;
@Test
public void userBySexTest() {
try {
List<User> userList = userFacade.getBySex("男");
userList.forEach(System.out::println);
} catch (RemoteException e) {
e.printStackTrace();
}
}
通過測(cè)試類可以看出,這和我們平時(shí)的程序調(diào)用內(nèi)部方法沒什么區(qū)別!
rmi調(diào)用過程
大家可以通過下面文章加深了解:
有兩個(gè)遠(yuǎn)程服務(wù)接口可供client調(diào)用馍管,F(xiàn)actory和Product接口
-
FactoryImpl類實(shí)現(xiàn)了Factory接口,ProductImpl類實(shí)現(xiàn)了Product接口
1. FactoryImpl被注冊(cè)到了rmi-registry中 2. client端請(qǐng)求一個(gè)Factory的引用 3. rmi-registry返回client端一個(gè)FactoryImpl的引用 4. client端調(diào)用FactoryImpl的遠(yuǎn)程方法請(qǐng)求一個(gè)ProductImpl的遠(yuǎn)程引用 5. FactoryImpl返回給client端一個(gè)ProductImpl引用 6. client通過ProductImpl引用調(diào)用遠(yuǎn)程方法
socket工廠文檔:
https://docs.oracle.com/javase/8/docs/technotes/guides/rmi/socketfactory/index.html
Zookeeper實(shí)現(xiàn)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 目錄下薪韩,啟動(dòng) Zookeeper:
cd zookeeper-3.4.12/bin
./zkServer.sh start
消費(fèi)者:
public class RmiConsumer {
// 用于等待 SyncConnected 事件觸發(fā)后繼續(xù)執(zhí)行當(dāng)前線程
private CountDownLatch latch = new CountDownLatch(1);
// 定義一個(gè) volatile 成員變量确沸,用于保存最新的 RMI 地址(考慮到該變量或許會(huì)被其它線程所修改,一旦修改后俘陷,該變量的值會(huì)影響到所有線程)
private volatile List<String> urlList = new ArrayList<>();
// 構(gòu)造器
public RmiConsumer() {
ZooKeeper zk = connectServer(); // 連接 ZooKeeper 服務(wù)器并獲取 ZooKeeper 對(duì)象
if (zk != null) {
watchNode(zk); // 觀察 /registry 節(jié)點(diǎn)的所有子節(jié)點(diǎn)并更新 urlList 成員變量
}
}
// 查找 RMI 服務(wù)
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 中只有一個(gè)元素罗捎,則直接獲取該元素
log.debug("using only url: {}", url);
} else {
url = urlList.get(ThreadLocalRandom.current().nextInt(size)); // 若 urlList 中存在多個(gè)元素,則隨機(jī)獲取一個(gè)元素
log.debug("using random url: {}", url);
}
service = lookupService(url); // 從 JNDI 中查找 RMI 服務(wù)
}
return service;
}
// 連接 ZooKeeper 服務(wù)器
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(); // 喚醒當(dāng)前正在執(zhí)行的線程
}
}
});
latch.await(); // 使當(dāng)前線程處于等待狀態(tài)
} catch (IOException | InterruptedException e) {
log.error("", e);
}
return zk;
}
// 觀察 /registry 節(jié)點(diǎn)下所有子節(jié)點(diǎn)是否有變化
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); // 若子節(jié)點(diǎn)有變化拉盾,則重新調(diào)用該方法(為了獲取最新子節(jié)點(diǎn)中的數(shù)據(jù))
}
});
List<String> dataList = new ArrayList<>(); // 用于存放 /registry 所有子節(jié)點(diǎn)中的數(shù)據(jù)
for (String node : nodeList) {
byte[] data = zk.getData(Constant.ZK_REGISTRY_PATH + "/" + node, false, null); // 獲取 /registry 的子節(jié)點(diǎn)中的數(shù)據(jù)
dataList.add(new String(data));
}
log.debug("node data: {}", dataList);
urlList = dataList; // 更新最新的 RMI 地址
} catch (KeeperException | InterruptedException e) {
log.error("", e);
}
}
// 在 JNDI 中查找 RMI 遠(yuǎn)程服務(wù)對(duì)象
@SuppressWarnings("unchecked")
private <T> T lookupService(String url) {
T remote = null;
try {
remote = (T) Naming.lookup(url);
} catch (NotBoundException | MalformedURLException | RemoteException e) {
log.error("遠(yuǎn)程查找出錯(cuò)!", e);
}
return remote;
}
}
生產(chǎn)者:
public class RmiProvider {
/**
* 用于等待 SyncConnected 事件觸發(fā)后繼續(xù)執(zhí)行當(dāng)前線程
*/
private CountDownLatch latch = new CountDownLatch(1);
// 發(fā)布 RMI 服務(wù)并注冊(cè) RMI 地址到 ZooKeeper 中
public void publish(Remote remote, String host, int port) {
String url = publishService(remote, host, port); // 發(fā)布 RMI 服務(wù)并返回 RMI 地址
if (url != null) {
ZooKeeper zk = connectServer(); // 連接 ZooKeeper 服務(wù)器并獲取 ZooKeeper 對(duì)象
if (zk != null) {
createNode(zk, url); // 創(chuàng)建 ZNode 并將 RMI 地址放入 ZNode 上
}
}
}
/**
*發(fā)布 RMI 服務(wù)
*/
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 服務(wù)器
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(); // 喚醒當(dāng)前正在執(zhí)行的線程
}
}
});
latch.await(); // 使當(dāng)前線程處于等待狀態(tài)
} catch (IOException | InterruptedException e) {
log.error("", e);
}
return zk;
}
/**
* 創(chuàng)建節(jié)點(diǎn)
*/
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); // 創(chuàng)建一個(gè)臨時(shí)性且有序的 ZNode
log.debug("create zookeeper node ({} => {})", path, url);
} catch (KeeperException | InterruptedException e) {
log.error("", e);
}
}
}
圖解:
代碼已上傳到GitHub上:https://github.com/jayknoxqu/rmi-example