hadoop_rpc调用之client（二）

hadoop_rpc调用之client（2）
主要的几个类说明：
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.当出现异常时，直接关闭连接，并处理没有返回结果的call4.Connection中保存Call的调用，使用的是一个HashTable，没有进行数量限制，若服务器阻塞后，Call会积累，感觉这个时候是有问题的。

查看更多 下一篇