JAVA RMI线程模型及内部实现机制

 

1 RMI内部实现

  JAVA RMI是JAVA分布式结构的基础。 远程对象的通信过程中， RMI 使用标准机制： stub 和skeleton 。远程对象的 stub 担当远程对象的客户本地代表或代理人角色，调用程序将调 用本地stub 的方法，而本地 stub 将负责执行对远程对象的方法调用。在 RMI 中，远程对象的 stub 与该远程对象所实现的远程接口集相同。调用 stub 的方法时将执行下列操作： 
   (1) 初始化与包含远程对象的远程虚拟机的连接； 
　 (2) 对远程虚拟机的参数进行编组-传输参数； 
　 (3) 等待远程方法调用结果； 
　 (4) 解编（读取）返回值或返回的异常； 
　 (5) 将值返回给调用程序。

  为了向调用程序展示比较简单的调用机制， stub 将参数的序列化和网络级通信等细节隐藏了起来。在远程虚拟机中，每个远程对象都可以 有相应的 skeleton。s keleton 负责将调用分配给实际的远程对象实现。它在接收方法调用时 执行下列操作：

  (1) 解编（读取）远程方法的参数；

  (2) 调用实际远程对象实现上的方法；

  (3) 将结果（返回值或异常）编组（写入并传输）给调用程序。  

  stub 和 skeleton 由 rmic 编译器生成。在最新的JDK中，不需要手工生产stub和skeleton，用动态代理生成的Proxy代替了stub，而skeleton则取消了。

  我们可以查看源代码来了解RMI的内部实现。Server端调用UnicastRemoteObject的export方法输出远程对象，export方法会在一个线程里监听某个TCP端口上的方法调用请求：

 

public void exportObject(Target target) throws RemoteException {

  // other code

while (true) {
    ServerSocket myServer = server;
    if (myServer == null)
      return;
    Throwable acceptFailure = null;
    final Socket socket;

    try {
    socket = myServer.accept();

    /*
     * Find client host name (or "0.0.0.0" if unknown)
     */
    InetAddress clientAddr = socket.getInetAddress();
    String clientHost = (clientAddr != null
                 ? clientAddr.getHostAddress()
                 : "0.0.0.0");

    /*
     * Spawn non-system thread to handle the connection
     */
    Thread t = (Thread)
        java.security.AccessController.doPrivileged (
            new NewThreadAction(new ConnectionHandler(socket,
                              clientHost),
                    "TCP Connection(" + ++ threadNum +
                    ")-" + clientHost,
                    true, true));
    t.start();

    } catch (IOException e) {
    acceptFailure = e;
    } catch (RuntimeException e) {
    acceptFailure = e;
    } catch (Error e) {
    acceptFailure = e;
    }
}

// other code

 

  上面的代码已被修改以展示主要的要点，Server端就是在ServerSocket的accept方法上面监听到来的请求，如果有新的方法调用请求到来，Server产生一个单独的线程来处理新接收的请求：

 

public void dispatch(Remote obj, RemoteCall call) throws IOException {
// positive operation number in 1.1 stubs;
// negative version number in 1.2 stubs and beyond...
int num;
long op;
try {
    // read remote call header
    ObjectInput in;
    try {
    in = call.getInputStream();
    num = in.readInt();
    if (num >= 0) {
        if (skel != null) {
        oldDispatch(obj, call, num);
        return;
        } else {
        throw new UnmarshalException(
            "skeleton class not found but required " +
            "for client version");
        }
    }
    op = in.readLong();
    } catch (Exception readEx) {
    throw new UnmarshalException("error unmarshalling call header",
                     readEx);
    }
    /*
     * Since only system classes (with null class loaders) will be on
     * the execution stack during parameter unmarshalling for the 1.2
     * stub protocol, tell the MarshalInputStream not to bother trying
     * to resolve classes using its superclasses's default method of
     * consulting the first non-null class loader on the stack.
     */
    MarshalInputStream marshalStream = (MarshalInputStream) in;
    marshalStream.skipDefaultResolveClass();
    Method method = (Method) hashToMethod_Map.get(new Long(op));
    if (method == null) {
    throw new UnmarshalException("invalid method hash");
    }
    // if calls are being logged, write out object id and operation
    logCall(obj, method);
    // unmarshal parameters
    Class[] types = method.getParameterTypes();
    Object[] params = new Object[types.length];
    try {
    unmarshalCustomCallData(in);
    for (int i = 0; i < types.length; i++) {
        params[i] = unmarshalValue(types[i], in);
    }
    } catch (java.io.IOException e) {
    throw new UnmarshalException(
        "error unmarshalling arguments", e);
    } catch (ClassNotFoundException e) {
    throw new UnmarshalException(
        "error unmarshalling arguments", e);
    } finally {
    call.releaseInputStream();
    }
    // make upcall on remote object
    Object result;
    try {
    result = method.invoke(obj, params);
    } catch (InvocationTargetException e) {
    throw e.getTargetException();
    }
    // marshal return value
    try {
    ObjectOutput out = call.getResultStream(true);
    Class rtype = method.getReturnType();
    if (rtype != void.class) {
        marshalValue(rtype, result, out);
    }
    } catch (IOException ex) {
    throw new MarshalException("error marshalling return", ex);
    /*
     * This throw is problematic because when it is caught below,
     * we attempt to marshal it back to the client, but at this
     * point, a "normal return" has already been indicated,
     * so marshalling an exception will corrupt the stream.
     * This was the case with skeletons as well; there is no
     * immediately obvious solution without a protocol change.
     */
    }
} catch (Throwable e) {
    logCallException(e);

    ObjectOutput out = call.getResultStream(false);
    if (e instanceof Error) {
    e = new ServerError(
        "Error occurred in server thread", (Error) e);
    } else if (e instanceof RemoteException) {
    e = new ServerException(
        "RemoteException occurred in server thread",
        (Exception) e);
    }
    if (suppressStackTraces) {
    clearStackTraces(e);
    }
    out.writeObject(e);
} finally {
    call.releaseInputStream(); // in case skeleton doesn't
    call.releaseOutputStream();
}
   }
protected static void marshalValue(Class type, Object value,
                   ObjectOutput out)
throws IOException
   {
if (type.isPrimitive()) {
    if (type == int.class) {
    out.writeInt(((Integer) value).intValue());
    } else if (type == boolean.class) {
    out.writeBoolean(((Boolean) value).booleanValue());
    } else if (type == byte.class) {
    out.writeByte(((Byte) value).byteValue());
    } else if (type == char.class) {
    out.writeChar(((Character) value).charValue());
    } else if (type == short.class) {
    out.writeShort(((Short) value).shortValue());
    } else if (type == long.class) {
    out.writeLong(((Long) value).longValue());
    } else if (type == float.class) {
    out.writeFloat(((Float) value).floatValue());
    } else if (type == double.class) {
    out.writeDouble(((Double) value).doubleValue());
    } else {
    throw new Error("Unrecognized primitive type: " + type);
    }
} else {
    out.writeObject(value);
}
   }
rotected static Object unmarshalValue(Class type, ObjectInput in)
throws IOException, ClassNotFoundException
   {
if (type.isPrimitive()) {
    if (type == int.class) {
    return new Integer(in.readInt());
    } else if (type == boolean.class) {
    return new Boolean(in.readBoolean());
    } else if (type == byte.class) {
    return new Byte(in.readByte());
    } else if (type == char.class) {
    return new Character(in.readChar());
    } else if (type == short.class) {
    return new Short(in.readShort());
    } else if (type == long.class) {
    return new Long(in.readLong());
    } else if (type == float.class) {
    return new Float(in.readFloat());
    } else if (type == double.class) {
    return new Double(in.readDouble());
    } else {
    throw new Error("Unrecognized primitive type: " + type);
    }
} else {
    return in.readObject();
}
   }

 

   dispatch方法处理接收的请求，先从输入流中读取方法的编号来获得方法名称：op = in.readLong()，然后读取方法的所有参数：params[i] = unmarshalValue(types[i], in)，接着就可以执行方法调用了：result = method.invoke(obj, params)，最后把方法执行结果写入到输出流中：marshalValue(rtype, result, out)。一个方法调用就执行完成了。

  Client端请求一个远程方法调用时，先建立连接：Connection conn = ref.getChannel().newConnection()，然后发送方法参数： marshalValue(types[i], params[i], out)，再发送执行方法请求：call.executeCall()，最后得到方法的执行结果：Object returnValue = unmarshalValue(rtype, in)，并关闭接：

ref.getChannel().free(conn, true)。

 

 public Object invoke(Remote obj,
         java.lang.reflect.Method method,
         Object[] params,
         long opnum)
throws Exception
   {
if (clientRefLog.isLoggable(Log.VERBOSE)) {
    clientRefLog.log(Log.VERBOSE, "method: " + method);
}
if (clientCallLog.isLoggable(Log.VERBOSE)) {
    logClientCall(obj, method);
}

Connection conn = ref.getChannel().newConnection();
RemoteCall call = null;
boolean reuse = true;
/* If the call connection is "reused" early, remember not to
 * reuse again.
 */
boolean alreadyFreed = false;
try {
    if (clientRefLog.isLoggable(Log.VERBOSE)) {
    clientRefLog.log(Log.VERBOSE, "opnum = " + opnum);
    }
    // create call context
    call = new StreamRemoteCall(conn, ref.getObjID(), -1, opnum);
    // marshal parameters
    try {
    ObjectOutput out = call.getOutputStream();
    marshalCustomCallData(out);
    Class[] types = method.getParameterTypes();
    for (int i = 0; i < types.length; i++) {
        marshalValue(types[i], params[i], out);
    }
    } catch (IOException e) {
    clientRefLog.log(Log.BRIEF,
        "IOException marshalling arguments: ", e);
    throw new MarshalException("error marshalling arguments", e);
    }
    // unmarshal return
    call.executeCall();
    try {
    Class rtype = method.getReturnType();
    if (rtype == void.class)
        return null;
    ObjectInput in = call.getInputStream();

    /* StreamRemoteCall.done() does not actually make use
     * of conn, therefore it is safe to reuse this
     * connection before the dirty call is sent for
     * registered refs.
     */
    Object returnValue = unmarshalValue(rtype, in);
    /* we are freeing the connection now, do not free
     * again or reuse.
     */
    alreadyFreed = true;
    /* if we got to this point, reuse must have been true. */
    clientRefLog.log(Log.BRIEF, "free connection (reuse = true)");
    /* Free the call's connection early. */
    ref.getChannel().free(conn, true);
    return returnValue;

    } catch (IOException e) {
    clientRefLog.log(Log.BRIEF,
             "IOException unmarshalling return: ", e);
    throw new UnmarshalException("error unmarshalling return", e);
    } catch (ClassNotFoundException e) {
    clientRefLog.log(Log.BRIEF,
        "ClassNotFoundException unmarshalling return: ", e);
    throw new UnmarshalException("error unmarshalling return", e);
    } finally {
    try {
        call.done();
    } catch (IOException e) {
        /* WARNING: If the conn has been reused early,
         * then it is too late to recover from thrown
         * IOExceptions caught here. This code is relying
         * on StreamRemoteCall.done() not actually
         * throwing IOExceptions.
         */
        reuse = false;
    }
    }
} catch (RuntimeException e) {
    /*
     * Need to distinguish between client (generated by the
     * invoke method itself) and server RuntimeExceptions.
     * Client side RuntimeExceptions are likely to have
     * corrupted the call connection and those from the server
     * are not likely to have done so.  If the exception came
     * from the server the call connection should be reused.
     */
    if ((call == null) ||
    (((StreamRemoteCall) call).getServerException() != e))
           {
    reuse = false;
    }
    throw e;
} catch (RemoteException e) {
    /*
     * Some failure during call; assume connection cannot
     * be reused.  Must assume failure even if ServerException
     * or ServerError occurs since these failures can happen
     * during parameter deserialization which would leave
     * the connection in a corrupted state.
     */
    reuse = false;
    throw e;
} catch (Error e) {
    /* If errors occurred, the connection is most likely not
            *  reusable.
     */
    reuse = false;
    throw e;
} finally {
    /* alreadyFreed ensures that we do not log a reuse that
     * may have already happened.
     */
    if (!alreadyFreed) {
    if (clientRefLog.isLoggable(Log.BRIEF)) {
        clientRefLog.log(Log.BRIEF, "free connection (reuse = " +
                   reuse + ")");
    }
    ref.getChannel().free(conn, reuse);
    }
}
   }

 

2 RMI线程模型

  在JDK1.5及以前版本中，RMI每接收一个远程方法调用就生成一个单独的线程来处理这个请求，请求处理完成后，这个线程就会释放：

 

Thread t = (Thread)
            java.security.AccessController.doPrivileged (
                new NewThreadAction(new ConnectionHandler(socket,
                                  clientHost),
                        "TCP Connection(" + ++ threadNum +
                        ")-" + clientHost,
                        true, true));

 

   在JDK1.6之后，RMI使用线程池来处理新接收的远程方法调用请求-ThreadPoolExecutor。

  下面是一个简单的RMI程序的执行线程抓图，我们可以更好的了解RMI的线程机制。这个简单的RMI程序是服务端有一个远程方法实现，一个客户端同时请求执行这个远程方法100次。在JDK1.5中执行时生成的线程如下图所示，每个方法调用请求都是在一个单独的线程里执行，即 A Thread per Request。

在JDK1.6中执行时生成的线程如下图所示，这些线程都是在ThreadPoolExecutor线程池中执行的。

3 RMI线程池参数

  在JDK1.6中，RMI提供了可配置的线程池参数属性：

sun.rmi.transport.tcp.maxConnectionThread - 线程池中的最大线程数量

sun.rmi.transport.tcp.threadKeepAliveTime - 线程池中空闲的线程存活时间

 

转自：http://blog.csdn.net/sureyonder/article/details/5653609