本文主要记录hadoop rpc的客户端部分的初始化和调用的过程，下面的介绍中主要通过DFSClient来说明，为什么用DFSClient呢？DFSClient作为namenode的客户端，通过rpc来操作hdfs。限于篇幅，本文对下文引用到的类，做了较大的剪裁，只给出了关键的部分，如有疑问，可以一起交流。

DFSClient的初始化

DFSClient的初始化主要看其构造函数，其中rpc部分我们主要关注属性final ClientProtocol namenode，DFSClient的文件系统操作都是由他代理完成，构造函数中的关键代码如下：

public DFSClient(URI nameNodeUri, ClientProtocol rpcNamenode,Configuration conf, FileSystem.Statistics stats)throws IOException {proxyInfo = NameNodeProxies.createProxy(conf, nameNodeUri,ClientProtocol.class);this.dtService = proxyInfo.getDelegationTokenService();this.namenode = proxyInfo.getProxy();
}

显然，DFSClient中的namenode是一个代理类。

接着NameNodeProxies类的createProxy方法，下面给出了NameNodeProxies中需要用到的一些方法：

public class NameNodeProxies {
public static <T> ProxyAndInfo<T> createProxy(Configuration conf,URI nameNodeUri, Class<T> xface) throws IOException {return createNonHAProxy(conf, NameNode.getAddress(nameNodeUri), xface,UserGroupInformation.getCurrentUser(), true);
}public static <T> ProxyAndInfo<T> createNonHAProxy(Configuration conf, InetSocketAddress nnAddr, Class<T> xface,UserGroupInformation ugi, boolean withRetries) throws IOException {proxy = (T) createNNProxyWithClientProtocol(nnAddr, conf, ugi,withRetries);return new ProxyAndInfo<T>(proxy, dtService);
}/**这部分是重点
*/
private static ClientProtocol createNNProxyWithClientProtocol(InetSocketAddress address, Configuration conf, UserGroupInformation ugi,boolean withRetries) throws IOException {ClientNamenodeProtocolPB proxy = RPC.getProtocolProxy(ClientNamenodeProtocolPB.class, version, address, ugi, conf,NetUtils.getDefaultSocketFactory(conf),org.apache.hadoop.ipc.Client.getTimeout(conf), defaultPolicy).getProxy();proxy = (ClientNamenodeProtocolPB) RetryProxy.create(ClientNamenodeProtocolPB.class,new DefaultFailoverProxyProvider<ClientNamenodeProtocolPB>(ClientNamenodeProtocolPB.class, proxy),methodNameToPolicyMap,defaultPolicy);return new ClientNamenodeProtocolTranslatorPB(proxy);
}
}

该类中前面两个方法做跳转用，直接看createNNProxyWithClientProtocol方法，这里两行很关键的代码，proxy实例的初始化，这里先提示注意前一行中的getProxy() 对于这个方法是需要注意的，这样也保证了类型的一致。

这时候就不得不调出RPC这个类来看看他是怎么生成proxy的实例的了，看代码：ProtobufRpcEngineProtobufRpcEngineProtobufRpcEngineProtobufRpcEngine

public class RPC {
public static <T> ProtocolProxy<T> getProtocolProxy(Class<T> protocol,long clientVersion,InetSocketAddress addr,UserGroupInformation ticket,Configuration conf,SocketFactory factory,int rpcTimeout,RetryPolicy connectionRetryPolicy) throws IOException {    if (UserGroupInformation.isSecurityEnabled()) {SaslRpcServer.init(conf);}return getProtocolEngine(protocol,conf).getProxy(protocol, clientVersion,addr, ticket, conf, factory, rpcTimeout, connectionRetryPolicy);}
}

RPC中还是需要进一步的跳转，但是这里需要注意，getProtocolEngine这个方法，这里做一个说明，查看
RpcEngine的依赖，看图：  在我的2.4.1的hadoop的版本中，hadoop的序列化框架已经用了Protobuf，所以getProtocolEngine方法得到的是ProtobufRpcEngine类的一个实例，那好，我们进一步跟踪ProtobufRpcEngine类的getProxy方法，看代码：

public class ProtobufRpcEngine implements RpcEngine {public <T> ProtocolProxy<T> getProxy(Class<T> protocol, long clientVersion,InetSocketAddress addr, UserGroupInformation ticket, Configuration conf,SocketFactory factory, int rpcTimeout, RetryPolicy connectionRetryPolicy) throws IOException {final Invoker invoker = new Invoker(protocol, addr, ticket, conf, factory,rpcTimeout, connectionRetryPolicy);return new ProtocolProxy<T>(protocol, (T) Proxy.newProxyInstance(protocol.getClassLoader(), new Class[]{protocol}, invoker), false);}
}

对java的动态代理有点了解的人看到Proxy.newProxyInstance这个方法应该都很清楚这就是生成一个远程代理类实例（特别注意：在NameNodeProxies类的createNNProxyWithClientProtocol方法中getProxy方法拿到的对象也就是这个对象），其中的invoker参数，确实我们不能忽略的，因为他暗藏玄机，java的动态代理中，invoker的类需要实现InvocationHandler接口，该接口只听过一个方法invoke，共代理类使用，及通过Proxy.newProxyInstance生成的代理类，在使用的时候是通过InvocationHandler的invoke方法来起作用的。好吧，现在我们可以顺便看看在ProtobufRpcEngine类的getProxy方法中invoker局部变量的类依赖图：，显然有刚才提到的实现关系，现在再让我们看看Invoker的内部，包括构造函数和invoke方法：

private Invoker(Class<?> protocol, Client.ConnectionId connId,Configuration conf, SocketFactory factory) {this.remoteId = connId;this.client = CLIENTS.getClient(conf, factory, RpcResponseWrapper.class);this.protocolName = RPC.getProtocolName(protocol);this.clientProtocolVersion = RPC.getProtocolVersion(protocol);}public Object invoke(Object proxy, Method method, Object[] args)throws ServiceException {val = (RpcResponseWrapper) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,new RpcRequestWrapper(rpcRequestHeader, theRequest), remoteId);
}

在构造函数请注意一个属性client，他的类型正式 org.apache.hadoop.ipc.Client，而且在invoke方法中发起远程调用的正是这个client属性，能够读到这里的同学，相信应该比较清楚了，在DFSClient中发起远程访问的就是这个Client类的实例。

关于DFSClient的初始化阶段中关于rpc的部分，总结一句，就是创建一个namenode的代理对象，供后续的文件系统操作调用。

DFSClient的getFileLinkInfo方法

DFSClient提供了相当丰富的API供客户端操作hadoop的文件系统，这里以 getFileLinkInfo为例，讲解rpc客户端的调用过程。注意：如果是FileSystem类的话，请使用方法getFileLinkStatus，他对DFSClient提供的getFileLinkInfo做了一层包装，仅此而已。

直接看DFSClient中的代码：

public HdfsFileStatus getFileLinkInfo(String src) throws IOException {checkOpen();try {return namenode.getFileLinkInfo(src);} catch(RemoteException re) {throw re.unwrapRemoteException(AccessControlException.class,UnresolvedPathException.class);}}

很简答的一行代码，通过namenode属性的调用操作完成，看了DFSClient的初始化过程，我们很容易知道namenode的实例化类是ClientNamenodeProtocolTranslatorPB，继续看调用过程，代码转到了ClientNamenodeProtocolTranslatorPB中：

@Overridepublic HdfsFileStatus getFileLinkInfo(String src)throws AccessControlException, UnresolvedLinkException, IOException {GetFileLinkInfoRequestProto req = GetFileLinkInfoRequestProto.newBuilder().setSrc(src).build();try {GetFileLinkInfoResponseProto result = rpcProxy.getFileLinkInfo(null, req);return result.hasFs() ?  PBHelper.convert(rpcProxy.getFileLinkInfo(null, req).getFs()) : null;} catch (ServiceException e) {throw ProtobufHelper.getRemoteException(e);}}

这时候我们会发现一个属性rpcProxy，再回过头看看NameNodeProxies类的createProxy方法，我们就可以很清楚的知道，rpcProxy就是那个能发起远程调用的代理类，它封装了Invoker对象，当然就也有了使用Client类的能力，很好，这里我们稍微总结下，在DFSClient类中，调用getFileLinkInfo方法，最终就是通过Client的call方法，发起远程访问，获取数据。

这时候，我们可以进一步来探讨下Hadoop中RPC的Client类了，下面我把Client类主要的部分抽取出来了，看下面的代码：

public class Client {
Call createCall(RPC.RpcKind rpcKind, Writable rpcRequest) {return new Call(rpcKind, rpcRequest);}public Writable call(RPC.RpcKind rpcKind, Writable rpcRequest,ConnectionId remoteId, int serviceClass) throws IOException {final Call call = createCall(rpcKind, rpcRequest);Connection connection = getConnection(remoteId, call, serviceClass);connection.sendRpcRequest(call);                 // send the rpc requestreturn call.getRpcResponse();
}private class Connection extends Thread {private void receiveRpcResponse() {}public void sendRpcRequest(final Call call)throws InterruptedException, IOException {}
}
}

看了DFSclient的初始化部分，我们就可以知道，DFSClient的远程调用，是通过Client的call方法起作用的。其实Client的call方法已经很能够说明问题了，先封装一个call，然后获取连接，再得到结果。简单的说Client就是这样了。可以在稍微复杂一点，在Client的call方法中，封装了call后，getConnection的方法不仅是获取一个连接，同时会启动连接代表的线程，这个线程的作用就是等待请求的完成，完成后，将结果写到call中（该过程天内各国Connection的receiveRpcRespoce方法完成），在call方法中获取连接后，会发送请求的参数到namenode的服务端，等待namenode处理完毕，Connection的receiveRpcRespoce方法写返回结果，最后call方法中返回结果。大概的过程就是这个样子了。

好像整个过程也不太复杂，只是不熟悉的情况下跟踪代码会比较累点。