Hadoop RPC 实例

Datanode要向Namenode发送请求，希望得到一个字符串应答。假设Namenode的地址是127.0.0.1:9000

一、配置文件

org.apache.hadoop.conf. Configuration；对应的配置文件是core-site.xml。配置项以key-value的形式存储。

<configuration>
<!-- namenode -->
<property>
<name>dfs.namenode.adress</name>
<value>localhost:9000</value>
</property>
<property>
<name>dfs.namenode.handler.count</name>
<value>5</value>
</property>
</configuration>

二、用户需要自己定义一个协议接口来封装所有客户端向服务端发送的请求，该接口必须继承自org.apache.hadoop.ipc.VersionedProtocol。接口中的versionID用来标识不同的协议。

package rPCTest;
import org.apache.hadoop.ipc.VersionedProtocol;
public interface Datanode2NamenodeProtocil extends VersionedProtocol{
public static final long versionID = 19L;
public String getString();
}

三、Namenode端。调用方需要getServer得到一个RPC服务端。指定协议的实例，在指定的地址和端口上启动服务。

package rPCTest;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.Server;
import org.apache.hadoop.net.NetUtils;
public class Namenode implements Datanode2NamenodeProtocil {
private Configuration conf;
private Server rpcServer;
@Override
public String getString() {
return "This is a remote call......";
}
@Override
public long getProtocolVersion(String arg0, long arg1) throws IOException {
return Datanode2NamenodeProtocil.versionID;
}
public Namenode() throws IOException {
conf = new Configuration();
InetSocketAddress socAddr = NetUtils.createSocketAddr(conf.get(
"dfs.namenode.adress", ""));
int handlerCount = conf.getInt("dfs.namenode.handler.count", 5);
rpcServer = RPC.getServer(this, socAddr.getHostName(), socAddr
.getPort(), handlerCount, false, conf);
}
private void start() throws IOException {
rpcServer.start();
}

private void jion() throws InterruptedException {
rpcServer.join();//while()wait();
}
public static void main(String[] args) throws Exception {
Namenode namenode = new Namenode();
if (namenode != null){
namenode.start();
namenode.jion();
}
}
}

四、Datanode getProxy得到一个RPC客户端。getProxy方法会使用动态代理，创建一个指定服务端的代理。一个远程对象需要建立一个代理。

package rPCTest;
import java.io.IOException;
import java.net.InetSocketAddress;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
public class Datanode {
private Configuration conf;
private Datanode2NamenodeProtocil namenode;
public Datanode() throws IOException {
conf = new Configuration();
// connect to name node
InetSocketAddress nameNodeAddr = NetUtils.createSocketAddr(conf.get(
"dfs.namenode.adress", ""));
namenode = (Datanode2NamenodeProtocil) RPC.waitForProxy(
Datanode2NamenodeProtocil.class,
Datanode2NamenodeProtocil.versionID, nameNodeAddr, conf);
}
public String getString() {
return namenode.getString();
}
public void stop() {
RPC.stopProxy(namenode);
}
public static void main(String[] args) throws Exception {
Datanode datanode = new Datanode();
if (datanode != null)
System.out.println(datanode.getString());
datanode.stop();
}
}

五、运行Namenode



六、运行Datanode

This is a remote call......