hadoop rpc客户端初始化和调用过程怎么实现
这篇文章主要介绍“hadoop rpc客户端初始化和调用过程怎么实现”，在日常操作中，相信很多人在hadoop rpc客户端初始化和调用过程怎么实现问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”hadoop rpc客户端初始化和调用过程怎么实现”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
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中：
@Override
public 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 request
return 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方法中返回结果。大概的过程就是这个样子了。
到此，关于“hadoop rpc客户端初始化和调用过程怎么实现”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注网站，小编会继续努力为大家带来更多实用的文章！