`
jimmee
  • 浏览: 530346 次
  • 性别: Icon_minigender_1
  • 来自: 杭州
社区版块
存档分类
最新评论

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. 当出现异常时,直接关闭连接,并处理没有返回结果的call
  • 4. Connection中保存Call的调用,使用的是一个HashTable,没有进行数量限制,若服务器阻塞后,Call会积累,感觉这个时候是有问题的。

分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics