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客户端初始化和调用过程怎么实现”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注亿速云网站，小编会继续努力为大家带来更多实用的文章！