分布式系列五: RMI通信
2019-08-05

RPC(Remote Procedure Call)协议

RPC协议是一种通过网络从远程计算机上请求服务, 而不需要了解底层网络技术的协议, 在OSI模型中处在应用层和网络层.

作为一个规范, 使用RPC协议的框架有很多, Dubbo,Hessian等均使用这个协议, RMI也使用该协议实现.

RMI(Remote Method Invocation) 远程方法调用

RMI使用Java远程消息交换协议JRMP(Java Remote Messaging Protocol)进行通信,JRMP是纯java的.

    定义接口, 使其extends Remote接口, 方法需要抛出异常RemoteException, Remote是一个标记接口

public interface IRmiTest extends Remote { String hello() throws RemoteException;}

    实现接口, 使其extends UnicastRemoteObject, 需要有构造方法, 并抛出异常RemoteException

public class RmiTest extends UnicastRemoteObject implements IRmiTest { public RmiTest() throws RemoteException { } @Override public String hello() { return "Hello ...."; }}

    定义服务端, 注册和绑定

public class TestServer { public static void main(String[] args) throws RemoteException, AlreadyBoundException, MalformedURLException { IRmiTest rmiTest = new RmiTest(); LocateRegistry.createRegistry(8888); Naming.bind("rmi://localhost:8888/hello",rmiTest); System.out.println("server started"); }}

    定义客户端, lookup方法的参数url与服务端bind的必须一致. 接口需要定义为与服务端一致.

public class TestClient { public static void main(String[] args) throws RemoteException, MalformedURLException, NotBoundException { IRmiTest rmiTest = (IRmiTest) Naming.lookup("rmi://localhost:8888/hello"); System.out.println(rmiTest.hello()); }}

RMI实现机制

RMI屏蔽了底层复杂的网络调用, 使得远程对象的方法调用变得透明, 就像调用本地方法一样方便.下面深入探究下jdk中rmi的实现原理, 看看底层是如何实现远程调用的.首先, 需要了解下比较重要的两个角色stub和skeleton, 这两个角色封装了与网络相关的代码. 原始的交互式这样的,客户端--网络--服务器--具体服务. 有了这两个角色之后的模型变为: 客户端--stub--网络--skeleton--服务器--服务.可以参考的图维基百科

下面来看源码...

一.实例化RegistryImpl,初始化

LocateRegistry.createRegistry(8888);这句代码启动了一个注册器(其中有个Map对象来存储名称和服务的映射,这个后面再细看)

public static Registry createRegistry(int port) throws RemoteException { return new RegistryImpl(port);}

这个方法实例化了一个RegistryImpl的实例,RegistryImpl实现了Registry.

public RegistryImpl(final int var1) throws RemoteException { if(var1 == 1099 && System.getSecurityManager() != null) { try { AccessController.doPrivileged(new PrivilegedExceptionAction() { public Void run() throws RemoteException { LiveRef var1x = new LiveRef(RegistryImpl.id, var1); RegistryImpl.this.setup(new UnicastServerRef(var1x)); return null; } }, (AccessControlContext)null, new Permission[]{new SocketPermission("localhost:" + var1, "listen,accept")}); } catch (PrivilegedActionException var3) { throw (RemoteException)var3.getException(); } } else { LiveRef var2 = new LiveRef(id, var1); this.setup(new UnicastServerRef(var2)); }}

两个分支最终都调用了setup()方法, 主要关注该方法.if分支中var1=1099是指默认端口并且存在安全管理器的时候不做校验, 这是为了性能考虑.

private void setup(UnicastServerRef var1) throws RemoteException { this.ref = var1; // UnicastServerRef继承了RemoteRef,this.ref的类型就是RemoteRef var1.exportObject(this, (Object)null, true); }

setup方法的参数是包装后的UnicastServerRef对象, UnicastServerRef继承了RemoteRef因此可以赋值给ref变量. 该方法将调用委托给UnicastServerRef的方法exportObject()如果是拿文章开头的代码进行调试, 会发现这个方法会走两次, 除了RegistryImpl, 还有一次是RmiTest也会走这个方法.不同的是RegistryImpl会走下面代码中的if(var5 instanceof RemoteStub)分支语句, 这个语句最终将生成一个Skeleton实例并设置给当前实例的域变量skel, 不过自jdk1.2之后skeleton就没什么用了.

public Remote exportObject(Remote var1, Object var2, boolean var3) throws RemoteException { Class var4 = var1.getClass(); Remote var5; try { var5 = Util.createProxy(var4, this.getClientRef(), this.forceStubUse); } catch (IllegalArgumentException var7) { throw new ExportException("remote object implements illegal remote interface", var7); } if(var5 instanceof RemoteStub) { // 生成Skeleton实例并设置给当前实例的域变量skel this.setSkeleton(var1); } Target var6 = new Target(var1, this, var5, this.ref.getObjID(), var3); this.ref.exportObject(var6); //ref是实例化UnicastServerRef的时候传入的 this.hashToMethod_Map = (Map)hashToMethod_Maps.get(var4); return var5;}

上面方法首先根据Remote的参数var1创建了一个代理对象var5, var1是RegistryImpl类的实例. 然后实例化一个Target的实例, 从参数可以看到,Target对象包含了几乎之前代码的所有对象.然后将这个对象作为参数,调用LiveRef实例ref的exportObject()方法.

二. 网络连接和对象传输

public void exportObject(Target var1) throws RemoteException { this.ep.exportObject(var1);}

接上一步, RemoteRef的方法最终委托给TCPEndpoint的同名方法(委托模式), 到此代码将控制权传递给传输层.

public void exportObject(Target var1) throws RemoteException { synchronized(this) { this.listen(); ++this.exportCount; } boolean var2 = false; boolean var12 = false; try { var12 = true; super.exportObject(var1); var2 = true; var12 = false; } finally { if (var12) { if (!var2) { synchronized(this) { this.decrementExportCount(); } } } } if (!var2) { synchronized(this) { this.decrementExportCount(); } } }

这个方法实现了网络通信, 首先linsten()启动了一个ServerSocket的线程,并开始监听端口. 然后调用父类的方法将Target对象暴露出去, 此时服务端的初始化就完成了.

三. 注册服务

Naming.bind("rmi://localhost:8888/hello",rmiTest); 完成名称和服务对象的绑定.

public static void bind(String name, Remote obj) throws AlreadyBoundException, java.net.MalformedURLException, RemoteException{ ParsedNamingURL parsed = parseURL(name); Registry registry = getRegistry(parsed); if (obj == null) throw new NullPointerException("cannot bind to null"); registry.bind(parsed.name, obj);}

上面代码Naming类, 调用的是注册器Registrybind()方法

public void bind(String var1, Remote var2) throws RemoteException, AlreadyBoundException, AccessException { Hashtable var3 = this.bindings; synchronized(this.bindings) { Remote var4 = (Remote)this.bindings.get(var1); if (var4 != null) { throw new AlreadyBoundException(var1); } else { this.bindings.put(var1, var2); } }}

注册使用的容器是一个HashTable, 最终服务的名称和服务会被注册到这个map容器中.

到此为止, 服务端的初始化完成. 首先实例化了一个实现Register注册器的实例, 通过层层组装, 最终生成一个Target对象, 其中包含了组装过程中生成的全部状态, 最后调用RemoteRef的方法将对象转交给传输层对象TCPEndpoint的实例, 最终由这个对象启动Socket开启通信连接. 注册服务是通过Naming的方法委托调用Register注册器的方法实现, 并将结果最终注册到Register域的map对象中.

四. 客户端远程调用

IRmiTest rmiTest = (IRmiTest) Naming.lookup("rmi://localhost:8888/hello"); 客户端通过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.bind()方法一样, 这里lookup()最终也会委托给Registry的实例. 这个实例的实现不是用的服务端的Register_Impl, 而是使用RegistryImpl_Stub, 下面代码是lookup()的实现, 可以看出这里封装了网络io的一些逻辑.

public Remote lookup(String var1) throws AccessException, NotBoundException, RemoteException { try { RemoteCall var2 = this.ref.newCall(this, operations, 2, 4905912898345647071L); try { ObjectOutput var3 = var2.getOutputStream(); var3.writeObject(var1); } catch (IOException var17) { throw new MarshalException("error marshalling arguments", var17); } this.ref.invoke(var2); Remote var22; try { ObjectInput var4 = var2.getInputStream(); var22 = (Remote)var4.readObject(); } catch (IOException var14) { throw new UnmarshalException("error unmarshalling return", var14); } catch (ClassNotFoundException var15) { throw new UnmarshalException("error unmarshalling return", var15); } finally { this.ref.done(var2); } return var22; } catch (RuntimeException var18) { throw var18; } catch (RemoteException var19) { throw var19; } catch (NotBoundException var20) { throw var20; } catch (Exception var21) { throw new UnexpectedException("undeclared checked exception", var21); }}

至此, 服务端和客户端的连接完成, 可以开始通信了.

RMI自JDK1.1就已经提供了, 它提供了Java语言自己的RPC调用方式, 虽然有些老旧, 但依然经典. 目前有很多跨语言的技术或框架, 如后来的WebService, 再到目前的netty,shrift等基本已经取代了这种原始的调用方式, 他们是非阻塞的,且还能跨语言调用. 但熟悉RMI的实现方式对了解分布式系统的通信的实现原理有很大帮助.