Hadoop中的RPC

时间:2021-05-13 00:19:19

基于hadoop2.6.4,RPC相关的实现位于hadoop-common这个project中hadoop-common-project/hadoop-common/src/main/java的包package org.apache.hadoop.ipc中

而在hadoop-common-project/hadoop-common/test/main/java的包package org.apache.hadoop.ipc中TestIPC类中有一个具体的实现

RPC摒弃了java自带的RMI,使用了自己的模型,主要由虚基类Server/Client/RPC三个构成

hadoop.ipc.RPC 实现了一种远程过程调用的框架,应用可以直接定义过程调用的协议接口和协议的server端实现,就可以直接通过RPC框架获得RPC server和client端的接口代理。hadoop.ipc.RPC 的实现利用了 hadoop.ipc.Server 和 hadoop.ipc.Client这两个类, 这两个类实现了网络中非常典型的Request-Response模式服务器和客户端框架。用户可以通过定义一个协议接口并实现出Request和Response类,以及Server端的抽象处理接口(Server.call()) 就可以实现出完整的服务器程序,而客户端程序只需要在创建hadoop.ipc.Client实体时,指定协议接口和网络相关参数,然后调用 call() 就可以发送请求并获取响应。hadoop.ipc.RPC类中有两个重要的函数getServer和getProxy,getServer通过接口协议实现的实体来获取真正的server,getProxy获取远程访问的本地代理,getServer在yarn中被bind()所封装。

    public Server build() throws IOException, HadoopIllegalArgumentException {
if (this.conf == null) {
throw new HadoopIllegalArgumentException("conf is not set");
}
if (this.protocol == null) {
throw new HadoopIllegalArgumentException("protocol is not set");
}
if (this.instance == null) {
throw new HadoopIllegalArgumentException("instance is not set");
} return getProtocolEngine(this.protocol, this.conf).getServer(
this.protocol, this.instance, this.bindAddress, this.port,
this.numHandlers, this.numReaders, this.queueSizePerHandler,
this.verbose, this.conf, this.secretManager, this.portRangeConfig);
}
}

在TestIPC这个类中,有关于ipc的实际使用例子

  private static class TestInvocationHandler implements RpcInvocationHandler {
private static int retry = 0;
private final Client client;
private final Server server;
private final int total; TestInvocationHandler(Client client, Server server, int total) {
this.client = client;
this.server = server;
this.total = total;
} @Override
public Object invoke(Object proxy, Method method, Object[] args)
throws Throwable {
LongWritable param = new LongWritable(RANDOM.nextLong());
LongWritable value = (LongWritable) client.call(param,
NetUtils.getConnectAddress(server), null, null, 0, conf);
if (retry++ < total) {
throw new IOException("Fake IOException");
} else {
return value;
}
}

参考:

http://weixiaolu.iteye.com/blog/1504898

http://langyu.iteye.com/blog/1183337

http://blog.csdn.net/colorant/article/details/50803226

http://blog.csdn.net/wind5shy/article/details/9070245

http://watter1985.iteye.com/blog/1698558