Java RMI

lobin

浏览: 425534 次
性别:
来自: 上海

最近访客更多访客>>

jAmEs_

wangyy

sindyqiu

yezhouzdzy

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Java RMI是Java提供的远程方法调用，也就是RPC。

Java RMI在Registry上将远程对象存根（可能是静态存根，也可能是动态存根：Java动态代理对象，动态代理方式）和对应的绑定名进行关联，查找的时候根据绑定的名称进行查找。

Java RMI会将远程对象存根（可能是静态存根，也可能是动态存根：Java动态代理对象，动态代理方式）序列化存储在Registry上，在Registry上绑定名关联，查找的时候根据绑定的名称进行查找，找到后在序列化传给客户端，客户端通过从Registry上获取到的对象存根进行方法调用。

Java RMI的几个核心组件：

1、RMI服务接口组件

2、RMI服务实现组件

3、Stub组件

4、Skel组件

5、服务动态代理组件：动态生成的存根，也就是动态存根（Stub）

6、Registry

Java RMI（RPC）服务端远程对象调用服务骨架（Skeleton）

在很多远程方法调用（RPC）实现中，在开发的时候，通常会为远程对象生成对象的存根（Stub）和骨架（Skeleton），如在java1.5以前，开发RMI服务时，需要通过rmic生成对应的Stub和Skeleton。其他类似的技术如COM（DCOM）以及CORBA也会生成Stub和Skeleton。

Skeleton的作用是将用户的RPC请求转发给对应的远程对象调用。

Client ---<call>---> Stub ---<req>---> Skel ---<dispatch>----> ----<call>---->远程对象

虽然在java1.5及以后的版本rmic已经不再生成Skeleton，也内部也还是有Skeleton的。

Java针对RPC开发的RMI，为远程对象生成对象的Skeleton的接口定义如下：

public interface Skeleton {
    /**
     * Unmarshals arguments, calls the actual remote object implementation,
     * and marshals the return value or any exception.
     *
     * @param obj remote implementation to dispatch call to
     * @param theCall object representing remote call
     * @param opnum operation number
     * @param hash stub/skeleton interface hash
     * @exception java.lang.Exception if a general exception occurs.
     * @since JDK1.1
     * @deprecated no replacement
     */
    @Deprecated
    void dispatch(Remote obj, RemoteCall theCall, int opnum, long hash)
        throws Exception;

    /**
     * Returns the operations supported by the skeleton.
     * @return operations supported by skeleton
     * @since JDK1.1
     * @deprecated no replacement
     */
    @Deprecated
    Operation[] getOperations();
}

要实现一个基本的RPC远程调用，其实不需要Registry的。对于Java RMI，其实也可以不经过Registry，Stub代理调用请求远程对象时，底层transport层直接和远程服务建立连接，也可以实现远程方法调用：

public static void main(String[] args) throws Exception {
    // 2464817129585149033
    // -586335541     1550505784140     -32763
    FPingTaskService fPingTaskService2 = (FPingTaskService) Naming.lookup("rmi://:10999/FPingTaskService");
    int result2 = fPingTaskService2.fping(1234);
    System.out.println(result2);

    FPingTaskService fPingTaskService = getFPingTaskServiceStub();
    int result = fPingTaskService.fping(1234);
    System.out.println(result);
}

上面是经过Registry，从Registry上查找（发现）服务对象。下面那段代码是不经过Registry的。

private static FPingTaskService getFPingTaskServiceStub() throws ServiceException, IllegalAccessException, InvocationTargetException, InstantiationException {
    String cn = "FPingTaskServiceImpl" + "_Stub2";
    Class<?> c = null;
    try {
        c = Class.forName(cn);
    } catch (ClassNotFoundException e) {
        throw new ServiceException(e);
    }

    Class uidc = UID.class;
    Constructor<?> constructor = null;
    try {
        constructor = uidc.getDeclaredConstructor(new Class[] {int.class, long.class, short.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    UID uid = (UID) constructor.newInstance(-586335541, 1550505784140L, (short) -32763);


    Class oidc = ObjID.class;
    try {
        constructor = oidc.getDeclaredConstructor(new Class[] {long.class, UID.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    ObjID oid = (ObjID) constructor.newInstance(new Object[] {2464817129585149033L, uid});

//        ObjID oid = new ObjID(2);

    LiveRef ref = null;
    ref = new LiveRef(oid, 3576);
    RemoteRef remoteRef = new UnicastRef(ref);

    constructor = null;
    try {
        constructor = c.getDeclaredConstructor(RemoteRef.class);
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }

    FPingTaskService service = null;
    try {
        service = (FPingTaskService) constructor.newInstance(remoteRef);
    } catch (Exception e) {
        throw new ServiceException(e);
    }
    System.out.println();
    return service;
}

在以前老版本的java(1.2以前)，通过Stub/Skel来实现远程调用。开发RMI服务时，需要生成服务的Stub和Skel。现在的java版本不需要。

现在的java版本（1.2及以后）同时支持Stub/Skel的远程调用以及Java动态代理的远程调用。在1.5以前，会同时生成Stub和Skel，在1.5及以后，只生成Stub。

1.1 1.2 1.3 1.4 1.5 1.6 1.7（Java7） 1.8（Java8） Java9

Stub/Skel支持 Y Y Y Y ？？？？

Stub支持？？？？ Y Y Y Y

动态存根（Stub）支持 N N N N Y Y Y Y

生成Stub Y Y Y Y Y Y Y Y

生成Skel Y Y Y Y N N N N

Stub/Skel支持:

表示是否支持Stub/Skel方式的远程方法调用，是否会同时用到Stub和Skel。

Stub支持：

表示是否支持只使用Stub的远程方法调用，即只用到Stub，而不使用Skel。

动态存根（Stub）支持：

表示是否支持动态存根（Stub）方法的远程方法调用，即使用Java动态代理方法的远程方法调用。

生成Stub/Skel：

表示通过rmic编译时是否会生成Stub/Skel，或者会同时生成。

在1.2以前，是要生成服务的Stub和Skel，假设有一个服务a，对应的服务Stub和Skel：a_Stub，a_Skel，分别以_Stub，_Skel为后缀。

如果采用Stub/Skel方式进行远程调用，Stub运行在客户端，它作为远程服务的代理实现了服务接口的所有方法。Skel运行在服务端，它将客户端的远程方法调用请求转发给对应的服务实现。

Java RMI支持TCP，还支持HTTP：HTTP-wrapped的远程调用。

远程方法调用（RMI）

Export

Registry

绑定

发现（lookup）

关于Java RMI（RPC）的绑定（bind）、导出（export）以及发现（lookup）参考文章：https://lobin.iteye.com/blog/2437794

JRMP StreamProtocol

------------

transport协议

transport头部：

<magic,4 bytes,><version,2 bytes,>

magic值为：0x4a524d49，"JRMI"

version值为：2

如果采用的是HTTP：HTTP-wrapped的transport协议，需要在transport头部前面添加：

<"POST",4 bytes,0x504f5354, "POST"的ASCII码值>

body:

<protocol,2 bytes,>

protocol值为：0x4b，0x4c，0x4d

0x4b表示：stream protocol

0x4c表示：single operation per connection

0x4d表示：multiplex protocol

远程调用协议头部

<80(0x50),1 byte,><ObjID,n bytes,><op,4 bytes,><hash,8 bytes,>

ObjID -> <编号,8 bytes,><unique,4 bytes,><time,8 bytes,><count,2 bytes,>

op:操作编号字段

hash：

哈希值，表示调用的方法。参考sun.rmi.server.Util#computeMethodHash方法。哈希值的计算采用了SHA算法：

首先通过SHA算法生成哈希：

ByteArrayOutputStream var3 = new ByteArrayOutputStream(127);

MessageDigest var4 = MessageDigest.getInstance("SHA");
DataOutputStream var5 = new DataOutputStream(new DigestOutputStream(var3, var4));
String var6 = getMethodNameAndDescriptor(var0);
if(serverRefLog.isLoggable(Log.VERBOSE)) {
    serverRefLog.log(Log.VERBOSE, "string used for method hash: \"" + var6 + "\"");
}

var5.writeUTF(var6);
var5.flush();
byte[] var7 = var4.digest();

生成了生成SHA哈希后，经过如下操作将SHA返回的一个8为数组转化为一个长整型数：

long var1 = 0L;

for(int var8 = 0; var8 < Math.min(8, var7.length); ++var8) {
    var1 += (long)(var7[var8] & 255) << var8 * 8;
}

这样就计算出表示调用方法的哈希值。

方法签名：

方法名（参数类型（signature）列表）返回类型（signature）

如调用如下方法：

public void stats(int op)

则在计算方法的哈希值，在计算SHA哈希时所采用的方法签名为：

stats(I)V

如调用如下方法：

public String stats(int op, String ext)

则在计算方法的哈希值，在计算SHA哈希时所采用的方法签名为：

stats(ILjava/lang/String;)Ljava/lang/String;

ping协议头部:

<82,1 byte,>

DGC(distributed GC)协议头部:

<84,1 byte,><unique,4 bytes,><time,8 bytes,><count,2 bytes,>

Registry相关操作：

获取Registry

绑定（bind）

重新绑定（rebind）

解绑（unbind）

查找（lookup）

查询服务列表（list）

操作编号定义参考sun.rmi.registry.RegistryImpl_Skel中定义。

0: // bind(String, Remote)

1: // list()

2: // lookup(String)

表示lookup，如服务注册绑定的地址rmi://:1099/TestService，它绑定的名称为TestService，lookup的时候根据TestService到Registry上去lookup对应的远程服务对象。

3: // rebind(String, Remote)

4: // unbind(String)

body:

lookup：

body -> <绑定的服务名,n bytes,>

rmi://host:port/servicename，假设绑定的名称为TestService：

body -> <TestService,n bytes,>

远程方法调用相关操作：

如：

-1: 表示远程方法调用

不同op对应的body也不一样。

body:

RMI接口调用：

方法调用代码参考

通过动态存根方式（动态代理）：java.rmi.server.RemoteObjectInvocationHandler#invoke

通过静态存根方式（Stub）：如：FPingTaskServiceImpl_Stub#fping

body -> <参数列表,n bytes,>

<参数列表,n bytes,> -> <参数,n bytes,>...

<参数,n bytes,> -> 参考序列化机制。

几个关键接口：

RemoteCall

RemoteCall表示一个远程调用。该接口已废弃。

RemoteRef表示远程对象的句柄。

几个关键实现：

StreamRemoteCall

RemoteObjectInvocationHandler

创建服务代理代码如下：（sun.rmi.server.Util）

public static Remote createProxy(Class<?> var0, RemoteRef var1, boolean var2) throws StubNotFoundException {
    Class var3;
    try {
        var3 = getRemoteClass(var0);
    } catch (ClassNotFoundException var9) {
        throw new StubNotFoundException("object does not implement a remote interface: " + var0.getName());
    }

    if(var2 || !ignoreStubClasses && stubClassExists(var3)) {
        return createStub(var3, var1);
    } else {
        final ClassLoader var4 = var0.getClassLoader();
        final Class[] var5 = getRemoteInterfaces(var0);
        final RemoteObjectInvocationHandler var6 = new RemoteObjectInvocationHandler(var1);

        try {
	    return (Remote)AccessController.doPrivileged(new PrivilegedAction() {
	        public Remote run() {
		    return (Remote)Proxy.newProxyInstance(var4, var5, var6);
	        }
	    });
        } catch (IllegalArgumentException var8) {
	    throw new StubNotFoundException("unable to create proxy", var8);
        }
    }
}

通过Stub方式创建服务代理：

private static RemoteStub createStub(Class<?> var0, RemoteRef var1) throws StubNotFoundException {
    String var2 = var0.getName() + "_Stub";

    try {
        Class var3 = Class.forName(var2, false, var0.getClassLoader());
        Constructor var4 = var3.getConstructor(stubConsParamTypes);
        return (RemoteStub)var4.newInstance(new Object[]{var1});
    } catch (ClassNotFoundException var5) {
        throw new StubNotFoundException("Stub class not found: " + var2, var5);
    } catch (NoSuchMethodException var6) {
        throw new StubNotFoundException("Stub class missing constructor: " + var2, var6);
    } catch (InstantiationException var7) {
        throw new StubNotFoundException("Can\'t create instance of stub class: " + var2, var7);
    } catch (IllegalAccessException var8) {
        throw new StubNotFoundException("Stub class constructor not public: " + var2, var8);
    } catch (InvocationTargetException var9) {
        throw new StubNotFoundException("Exception creating instance of stub class: " + var2, var9);
    } catch (ClassCastException var10) {
        throw new StubNotFoundException("Stub class not instance of RemoteStub: " + var2, var10);
    }
}

通过Java动态代理创建服务代理：

final ClassLoader var4 = var0.getClassLoader();
final Class[] var5 = getRemoteInterfaces(var0);
final RemoteObjectInvocationHandler var6 = new RemoteObjectInvocationHandler(var1);

try {
    return (Remote)AccessController.doPrivileged(new PrivilegedAction() {
        public Remote run() {
            return (Remote)Proxy.newProxyInstance(var4, var5, var6);
        }
    });
} catch (IllegalArgumentException var8) {
    throw new StubNotFoundException("unable to create proxy", var8);
}

创建Skel：

static Skeleton createSkeleton(Remote var0) throws SkeletonNotFoundException {
    Class var1;
    try {
        var1 = getRemoteClass(var0.getClass());
    } catch (ClassNotFoundException var8) {
        throw new SkeletonNotFoundException("object does not implement a remote interface: " + var0.getClass().getName());
    }

    String var2 = var1.getName() + "_Skel";

    try {
        Class var3 = Class.forName(var2, false, var1.getClassLoader());
        return (Skeleton)var3.newInstance();
    } catch (ClassNotFoundException var4) {
        throw new SkeletonNotFoundException("Skeleton class not found: " + var2, var4);
    } catch (InstantiationException var5) {
        throw new SkeletonNotFoundException("Can\'t create skeleton: " + var2, var5);
    } catch (IllegalAccessException var6) {
        throw new SkeletonNotFoundException("No public constructor: " + var2, var6);
    } catch (ClassCastException var7) {
        throw new SkeletonNotFoundException("Skeleton not of correct class: " + var2, var7);
    }
}

在1.2以前，

创建服务代理：

public static RemoteStub getStub(String var0, Class var1, RemoteRef var2) throws StubNotFoundException {
    String var3 = var0 + "_Stub";
    RemoteStub var4 = null;

    try {
        Class var5 = loadClassFromClass(var3, var1);
        var4 = (RemoteStub)var5.newInstance();
    } catch (ClassNotFoundException var6) {
        throw new StubNotFoundException("Class not found: " + var3, var6);
    } catch (InstantiationException var7) {
        throw new StubNotFoundException("Can\'t create stub: " + var3, var7);
    } catch (IllegalAccessException var8) {
        throw new StubNotFoundException("No public constructor: " + var3, var8);
    } catch (ClassCastException var9) {
        throw new StubNotFoundException("Stub not of correct class: " + var3, var9);
    }

    RemoteStub.setRef(var4, var2);
    return var4;
}

public static RemoteStub getStub(String var0, RemoteRef var1) throws StubNotFoundException {
    return getStub(var0, (Class)null, var1);
}

public static RemoteStub getStub(Remote var0, RemoteRef var1) throws StubNotFoundException {
    Class var2 = null;

    try {
        var2 = getRemoteClass(var0);
        return getStub(var2.getName(), var2, var1);
    } catch (ClassNotFoundException var3) {
        throw new StubNotFoundException("Object does not implement an interface that extends java.rmi.Remote: " + var0.getClass().getName());
    }
}

public static RemoteStub getProxy(Remote var0) throws RemoteException {
    RemoteStub var2;
    if(var0 instanceof RemoteStub) {
        var2 = (RemoteStub)var0;
    } else {
        Target var1;
        if((var1 = ObjectTable.getTarget(var0)) == null) {
            throw new StubNotFoundException("Remote object not exported: " + var0.getClass().getName());
        }

        var2 = var1.getStub();
    }

    return var2;
}

创建服务的Skel，

创建Skel：

public static Skeleton getSkeleton(Remote var0) throws SkeletonNotFoundException {
    Class var1;
    try {
        var1 = getRemoteClass(var0);
    } catch (ClassNotFoundException var8) {
        throw new SkeletonNotFoundException("Object does not implement an interface that extends java.rmi.Remote: " + var0.getClass().getName());
    }

    String var2 = var1.getName() + "_Skel";

    try {
        Class var3 = loadClassFromClass(var2, var1);
        return (Skeleton)var3.newInstance();
    } catch (ClassNotFoundException var4) {
        throw new SkeletonNotFoundException("Skeleton class not found: " + var2, var4);
    } catch (InstantiationException var5) {
        throw new SkeletonNotFoundException("Can\'t create skeleton: " + var2, var5);
    } catch (IllegalAccessException var6) {
        throw new SkeletonNotFoundException("No public constructor: " + var2, var6);
    } catch (ClassCastException var7) {
        throw new SkeletonNotFoundException("Skeleton not of correct class: " + var2, var7);
    }
}

在获取Registry的时候也是一样。如果不忽略Stub（ignoreStubClasses）并且Stub类存在的话，也是通过Stub和Skel的方式创建了一个Registry代理，否则通过java动态代理的方式创建了一个Registry代理。

Registry本身也实现了remote接口。

LocateRegistry#getRegistry

return (Registry) Util.createProxy(RegistryImpl.class, ref, false);

Registry

Registry是一个注册中心，它跟我们的服务注册中心类似。描述为一个简单的远程对象注册中心（Registry），它提供一些方法，根据远程对象的绑定名称存取远程对象引用。

java.rmi.registry.Registry

实例

参考文章：https://lobin.iteye.com/blog/615013

Java RMI（RPC）协议分析实例：

通过RemoteRef来调用：

int result3 = fping(1234);
System.out.println(result3);

public static int fping(int op) throws ServiceException {
    Class uidc = UID.class;
    Constructor<?> constructor = null;
    try {
        constructor = uidc.getDeclaredConstructor(new Class[] {int.class, long.class, short.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    UID uid = null;
    try {
        uid = (UID) constructor.newInstance(-586335541, 1550505784140L, (short) -32763);
    } catch (Exception e) {
        throw new ServiceException(e);
    }


    Class oidc = ObjID.class;
    try {
        constructor = oidc.getDeclaredConstructor(new Class[] {long.class, UID.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    ObjID oid = null;
    try {
        oid = (ObjID) constructor.newInstance(new Object[]{2464817129585149033L, uid});
    } catch (Exception e) {
        throw new ServiceException(e);
    }

//        ObjID oid = new ObjID(2);

    LiveRef ref = null;
    ref = new LiveRef(oid, 3576);
    RemoteRef remoteRef = new UnicastRef(ref);

    FPingTaskService service = new FPingTaskService() {
        @Override
        public int fping(int target) throws RemoteException {
            return 0;
        }
    };

    Class c = service.getClass();
    Method method = null;
    try {
        method = c.getDeclaredMethod("fping", new Class[] {int.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    long hash = Util.computeMethodHash(method);
    Object result = null;
    try {
        result = remoteRef.invoke(service, method, new Object[] {1234}, hash);
    } catch (Exception e) {
        throw new ServiceException(e);
    }
    return ((Integer) result).intValue();
}

通过StreamRemoteCall调用：

int result4 = fping2(1234);
System.out.println(result4);

public static int fping2(int op) throws ServiceException {
    Class uidc = UID.class;
    Constructor<?> constructor = null;
    try {
        constructor = uidc.getDeclaredConstructor(new Class[] {int.class, long.class, short.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    UID uid = null;
    try {
        uid = (UID) constructor.newInstance(-586335541, 1550505784140L, (short) -32763);
    } catch (Exception e) {
        throw new ServiceException(e);
    }


    Class oidc = ObjID.class;
    try {
        constructor = oidc.getDeclaredConstructor(new Class[] {long.class, UID.class});
    } catch (NoSuchMethodException e) {
        throw new ServiceException(e);
    }
    constructor.setAccessible(true);
    ObjID oid = null;
    try {
        oid = (ObjID) constructor.newInstance(new Object[]{2464817129585149033L, uid});
    } catch (Exception e) {
        throw new ServiceException(e);
    }

//        ObjID oid = new ObjID(2);

    LiveRef ref = new LiveRef(oid, 3576);

    long hash = computeMethodHash("fping(I)I"); // method signature to hash

    Connection conn = null;
    try {
        conn = ref.getChannel().newConnection();
        StreamRemoteCall remoteCall = new StreamRemoteCall(conn, ref.getObjID(), -1, hash);

        ObjectOutput output = remoteCall.getOutputStream();
        output.writeInt(1234);

        remoteCall.executeCall();

        ObjectInput input = remoteCall.getInputStream();

        int result = input.readInt();

        return result;
    } catch (RemoteException e) {
        throw new ServiceException(e);
    } catch (IOException e) {
        throw new ServiceException(e);
    } catch (Exception e) {
        throw new ServiceException(e);
    } finally {
        try {
            ref.getChannel().free(conn, true);
        } catch (RemoteException e) {
            throw new ServiceException(e);
        }
    }
}

public static long computeMethodHash(String sig) {
    long hash = 0L;
    ByteArrayOutputStream byteStream = new ByteArrayOutputStream(127);

    try {
        MessageDigest md = MessageDigest.getInstance("SHA");
        DataOutputStream dataOut = new DataOutputStream(new DigestOutputStream(byteStream, md));

        dataOut.writeUTF(sig);
        dataOut.flush();
        byte[] data = md.digest();

        for(int var8 = 0; var8 < Math.min(8, data.length); ++var8) {
            hash += (long)(data[var8] & 255) << var8 * 8;
        }
    } catch (IOException e) {
        hash = -1L;
    } catch (NoSuchAlgorithmException e) {
        throw new SecurityException(e.getMessage());
    }

    return hash;
}

分享到：

基于Zookeeper的Registry（服务注册中心） ... | 关于os的一些整理：第二篇：2、主引导程序

2016-12-18 18:05
浏览 472
评论(0)
论坛回复 / 浏览 (0 / 806)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

Java RMI

评论

发表评论

相关推荐

最近访客 更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论

Java RMI

评论

发表评论

相关推荐

Android: 动画

Android: LayoutInflater

Android: 配置清单

Android: 位置服务

Android: 位置服务

Android: SQLite

Android: 容器视图

Android: 布局

Android: ContentProvider

Android: 视图

Android: Activity

Clojure

Clojure

多路复用

I/O多路复用

I/O多路复用

多路复用

libevent: 安装

libevent: http

libevent: 安装

最近访客更多访客>>