hadoop_rpc调用之client（2）

della0887

主要的几个类说明：

• 1.Call，表示一次rpc的调用请求
  
• 2.Connection，表示一个client与server之间的连接，一个连接一个线程启动
  
• 3.ConnectionId：连接的标记（包括server地址，协议，其他一些连接的配置项信息）
  
• 4.ParallelCall：实现并行调用的请求
  
• 5.ParallelResults：并行调用的执行结果
  

执行逻辑：
1

• .当要执行一个调用时，将call放到connectin的map中；同时将请求发送到connection的输出流中，之后返回，并不一直持有connection并等待结果，所以是异步的处理过程；
  
• 2.connection自身线程不停从server读取请求返回，服务器返回的结果中包含请求的id，因此根据id从map中找到对应的call，从而设置call的调用结果；
  

可以看到，client的端的调用是很简单。
可以简单的来看一下代码：
public Writable call(Writable param, ConnectionId remoteId)，此方法是调用入口，代码分析：
/**
* 创建一个call，并得到连接，之后在连接中保存call，之后向连接的输出流写入请求
* 并返回, 底层使用的是oio（即blocking io），采用什么样的io与异步消息机制没有
* 必然联系.
**/
Call call = new Call(param);
Connection connection = getConnection(remoteId, call);
connection.sendParam(call);  
// 接口是同步的，异步变同步的操作再这里
synchronized (call) {
while (!call.done) {
try {
call.wait();                           // wait for the result
} catch (InterruptedException ie) {
// save the fact that we were interrupted
interrupted = true;
}
}
……
}

对应的，可以看到Connection类的receiveResponse方法里处理从server里读到的结果：
        int id = in.readInt();                    // try to read an id
Call call = calls.get(id);
Writable value = ReflectionUtils.newInstance(valueClass, conf);
value.readFields(in);                 // read value
call.setValue(value);
calls.remove(id);
Call调用setValue方式，会执行notify操作.
备注：

• 1.一个Client会对应多个Connection，并且会对这些Connection进行缓存；
  
• 2.一个Connection对应一个线程，这主要是内网中调用，节点之间的连接量应该不会太多（我想太多时，估计一个连接一个线程时就有问题了）
  
• 3.当出现异常时，直接关闭连接，并处理没有返回结果的call
  
• 4.Connection中保存Call的调用，使用的是一个HashTable，没有进行数量限制，若服务器阻塞后，Call会积累，感觉这个时候是有问题的。