1.发送请求：query.getCPUStatus("Intel");

2.传送到:RPCInvoker.invoke(Object proxy, Method method, Object[]args)

其中method:publicabstract org.hadoopinternal.ipc.copy.CPUStatusorg.hadoopinternal.ipc.copy.Query.getCPUStatus(java.lang.String)

args:[Intel]

3.调用:     ObjectWritable value = (ObjectWritable)

client.call(newRPCInvocation(method,args), remoteId);

4.构建 RPCInvocation对象

publicRPCInvocation(Methodmethod, Object[] parameters) {

this.methodName=method.getName();

this.parameterClasses=method.getParameterTypes();

this.parameters=parameters;

}

其中

methodName：getCPUStatus

parameterClasses：[classjava.lang.String]

parameters：[Intel]

5.构建ClientCall对象

publicWritablecall(Writable param, ClientConnectionIdremoteId)

throwsInterruptedException,IOException {

ClientCall call = newClientCall(param,this);

ClientConnection connection = getConnection(remoteId,call);

connection.sendParam(call);

......

}

其中：

protectedClientCall(Writableparam, Client client) {

this.client=client;

this.param=param;

synchronized(client){

this.id=client.counter++;

}

}

其中：

client：org.hadoopinternal.ipc.copy.Client@5fae6db3

param： RPCInvocation对象

id：2

6.调用connection.sendParam(call)： 

publicvoidsendParam(ClientCallcall) {

if(shouldCloseConnection.get()){

return;

}

DataOutputBuffer d=null;

try{

synchronized(this.out){

d = newDataOutputBuffer();

d.writeInt(call.id);

call.param.write(d);

byte[]data = d.getData();

intdataLength =d.getLength();

out.writeInt(dataLength);     //first putthe data length

out.write(data,0, dataLength);//write thedata

out.flush();

}

} catch(IOExceptione) {

markClosed(e);

} finally{

//the bufferis just an in-memory buffer, but it is still polite to

// closeearly

IOUtils.closeStream(d);

}

}

先和Call的ID和RPCInvocation对象加在一起，先发送长度，最后发送ID+RPCInvocation对象

7.服务器端接收

读取长度

count= server.channelRead(channel,dataLengthBuffer);

dataLength=dataLengthBuffer.getInt(); （18）

读取数据

data=ByteBuffer.allocate(dataLength);

count= server.channelRead(channel,data);

processData(data.array());

privatevoidprocessData(byte[]buf) throwsIOException, InterruptedException{

DataInputStream dis = newDataInputStream(newByteArrayInputStream(buf));

intid =dis.readInt(); // try toread an id

Writable param =ReflectionUtils.newInstance(server.paramClass,server.conf);//read param

param.readFields(dis);

ServerCall call = newServerCall(id,param, this);

server.callQueue.put(call);// queue thecall; maybe blocked here

rpcCount++;// Incrementthe rpc count

}

8.构建ServerCall对象：

publicServerCall(intid, Writableparam, ServerConnection connection) {

this.id=id;

this.param=param;

this.connection=connection;

this.timestamp=System.currentTimeMillis();

this.response=null;

}

id:2

param:RPCInvocation对象

connection:新构建的ServerConnection对象

timestamp：当前时间戳

response：回应结果

9. 拿出ServerCall，调用相应方法 

finalServerCallcall = server.callQueue.take();

value= server.call(call.connection.protocol,call.param,call.timestamp);

10.反射调用真正方法

将param还原成RPCInvocation对象，拿出MethodName、ParameterClasses、Parameters调用

真正方法：

publicWritablecall(Class＜?＞ protocol, Writableparam, longreceivedTime)

throwsIOException{

try{

RPCInvocation call = (RPCInvocation)param;

Method method = protocol.getMethod(call.getMethodName(), call.getParameterClasses());

method.setAccessible(true);

Object value = method.invoke(instance,call.getParameters());

returnnewObjectWritable(method.getReturnType(),value);

} catch(InvocationTargetExceptione) { }

}

根据method的ReturnType和结果构建返回值。

11.将结果（rv）写入ServerCall的Response中 

voidsetupResponse(ByteArrayOutputStreamresponse, ServerCall call,

Status status, Writable rv, String errorClass, Stringerror)

throwsIOException{

response.reset();

DataOutputStream out = newDataOutputStream(response);

out.writeInt(call.id);// write callid

out.writeInt(status.state);// writestatus

if(status ==Status.SUCCESS){

rv.write(out);

} else{

WritableUtils.writeString(out, errorClass);

WritableUtils.writeString(out, error);

}

call.setResponse(ByteBuffer.wrap(response.toByteArray()));

}

先写ServerCall的Id，再写状态Status，再写结果（rv）

12.response发送到客户端

intnumBytes= server.channelWrite(channel,call.response);

13.客户端接收

privatevoidreceiveResponse(){

if(shouldCloseConnection.get()){

return;

}

lastActivity.set(System.currentTimeMillis());

try{

intid =in.readInt();                   // try toread an id

ClientCall call = calls.get(id);

intstate= in.readInt();    // read callstatus

if(state ==Status.SUCCESS.state){

Writable value =ReflectionUtils.newInstance(client.valueClass,client.conf);

value.readFields(in);                // readvalue

call.setValue(value);

calls.remove(id);

} elseif(state ==Status.ERROR.state){

call.setException(new IOException(WritableUtils.readString(in)));

calls.remove(id);

} elseif(state ==Status.FATAL.state){

markClosed(new IOException(WritableUtils.readString(in)));

}

} catch(IOExceptione) {

markClosed(e);

}

}

先读ID:2

再读state:0

再读返回结果value:OW[class=classorg.hadoopinternal.ipc.copy.CPUStatus,value=CPU: Intel Create atFri Mar 14 10:45:56 CST 2014]

其中：

client.valueClass：classorg.apache.hadoop.io.ObjectWritable