Haoop IPC与Java Proxy机制

关于Hadoop IPC机制的描述，网上已有丰富的资料可供参考，比如：http://blog.csdn.net/zhanglizhe_cool/article/details/5712456，但大都围绕Hadoop中Server、Client的通信方式进行说明。这篇文章，以Hadoop IPC中对Java Proxy机制的使用作为主题，描述其基本工作原理和实现方法。

Java Proxy

Java Proxy（代理）是Java反射机制的一种。JVM为Java的用户提供了这种方法，使得Java代码的编写者可以为已有的类（被代理类）定义一个代理类的封装，代理对象可以在最终调用被代理类对象方法的前后添加更多自定义的操作。而代理类对象的使用者可以像使用被代理类对象一样，调用被代理类的方法。

这种机制的好处，是代理对象为其使用者屏蔽了在最终调用到被代理对象方法前后所进行的诸如建立网络通信等细节，代理的使用者在调用代理类对象的方法时，表面上看就跟调用被代理类对象方法具有相同的形式。

Hadoop IPC与Java Proxy

Hadoop的各进程之间，采用其自行实现的IPC机制进行通信。以下以NameNode和DataNode之间的通信为例。

在Hadoop中，DataNode通过定时的向NameNode发送心跳，来进行信息交互。Hadoop的NameNode类中，有一个public方法sendHeartbeat。DataNode对象向NameNode发送心跳，就是通过调用远程执行的NameNode对象的这个方法，并获取该方法的返回值来实现的。IPC通信中，主要涉及通信的类是org.apache.hadoop.ipc.Server以及org.apache.hadoop.ipc.Client。就直观的感觉讲，DataNode在这个发心跳的操作中，应该是这样的流程，DataNode通过Client建立到NamenNode的链接，然后向这个链接发送心跳请求，NameNode对象接收到请求之后，触发自己的sendHeartHeat方法，方法完成之后，向Client回送返回值。把视点集中到DataNode，这个交互过程，DataNode中需要进行建立网络连接、发送数据、接收数据三个操作，它应该要调用Client的相应方法来完成所有这些。但，实际情况比这个要简洁得多，因为Hadoop引入了org.apache.hadoop.ipc.RPC这个类，而这个类使用Java Proxy方式为DataNode屏蔽了繁琐的网络建立过程：

DataNode中的包含这样一个成员属性：public DatanodeProtocol namenode = null。DataNode进程在启动时，这个属性会被初始化：
this.namenode = (DatanodeProtocol) RPC.waitForProxy(DatanodeProtocol.class,                       DatanodeProtocol.versionID,                       nameNodeAddr,                        conf);
在这之后，当DataNode要向NameNode发送心跳时，调用namenode.sendHeartbeat即可。在DataNode看来，它直接调用自己成员变量namenode的sendHeartbeat方法，就直接实现了对远程NameNode对象这个sendHeartbeat方法的调用，形式变得非常简洁。这里，RPC使用Java Proxy机制，实现了这个看似神奇的远程调用。

研究RPC.waitForProxy可以知道，它最终触发了Java Proxy调用：VersionedProtocol proxy =        (VersionedProtocol) Proxy.newProxyInstance(            protocol.getClassLoader(), new Class[] { protocol },            new Invoker(protocol, addr, ticket, conf, factory, rpcTimeout))。Proxy机制中，当通过newProxyInstance调用获取一个代理类对象之后，对这个代理类对象的方法调用，都会触发到newProxyInstance这个方法所传入的第三个参数所定义的对象（java.lang.reflect.InvocationHandler接口的扩展类对象)的invoke方法。

看看DataNode在建立代理对象时所传入的这个org.apache.hadoop.ipc.RCP.Invoker类实现：

private static class Invoker implements InvocationHandler {
private Client.ConnectionId remoteId;
private Client client;
private boolean isClosed = false;

public Invoker(Class<? extends VersionedProtocol> protocol,        InetSocketAddress     address, UserGroupInformation ticket,        Configuration conf, SocketFactory factory, int   rpcTimeout) throws IOException {
this.remoteId = Client.ConnectionId.getConnectionId(address, protocol,          ticket, rpcTimeout, conf);
this.client = CLIENTS.getClient(conf, factory);
}

public Object invoke(Object proxy, Method method, Object[] args)      throws Throwable {
final boolean logDebug = LOG.isDebugEnabled();
long startTime = 0;
if (logDebug) {
startTime = System.currentTimeMillis();
}

ObjectWritable value = (ObjectWritable)        client.call(new Invocation(method, args), remoteId);
if (logDebug) {
long callTime = System.currentTimeMillis() – startTime;
LOG.debug(“Call: ” + method.getName() + ” ” + callTime);
}
return value.get();
}

…
}

Invoker对象在构造的时候，就会使用Client的相应方法，建立到Server的连接。Invoker为 InvocationHandler的实现类，把它注册为Proxy对象的invoker之后，只要客户端直接调用Proxy对象的方法，就会触发Invoker的invoke方法，Invoker类就是在这个Invoke方法中，发起了到Server的连结，将Method名称及参数发送到Server。随后Server会反向解析Method及这些参数并最终调用到Server侧对象的相应方法。