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会积累,感觉这个时候是有问题的。