hadoop2.6源码解读之DFSClient方法调用的RPC流程讲解

DFSClient 是应用程序访问hdfs的主要入口类

其方法调用最终通过RPC通信触发服务端响应请求。

以 rename方法为例，介绍整个流程

 public void rename(String src, String dst, Options.Rename... options)
      throws IOException {
    checkOpen();
    try {
      namenode.rename2(src, dst, options);
    }
    ...
    }

最终调用namenode 的方法

接下来 看下 namenode 是怎么构造出来的

先介绍两个协议

ClientProtocol协议

该协议定义了和名字节点交互的所有方法（如rename等方法）

@InterfaceAudience.Private
@InterfaceStability.Evolving
@KerberosInfo(
    serverPrincipal = DFSConfigKeys.DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY)
@TokenInfo(DelegationTokenSelector.class)
public interface ClientProtocol {
...
}

由此可见 ClientProtocol 其实就是一个 interface 定义

它有两个重要的子类

ClientNamenodeProtocolTranslatorPB 用于 client 端，发送请求 NameNodeRpcServer 用于server端，处理请求

ClientNamenodeProtocolPB 协议

ClientNamenodeProtocol.proto 生成 ClientNamenodeProtocolProtos.java

ClientNamenodeProtocolProtos 定义了 ClientNamenodeProtocol.BlockingInterface 接口

ClientNamenodeProtocolPB 继承了 这个 BlockingInterface

public interface ClientNamenodeProtocolPB extends 
  ClientNamenodeProtocol.BlockingInterface {
}

ClientNamenodeProtocolPB 这个接口协议 用处 是将ClientProtocol 各种方法 进行 protobuf 序列化包装，以便RPC进行调用

DFSClient 构造函数中

this.namenode = proxyInfo.getProxy();

proxyInfo 的来历是

proxyInfo = NameNodeProxies.createProxy(conf, nameNodeUri,
          ClientProtocol.class, nnFallbackToSimpleAuth);

在非HA模式下 createProxy最终执行于

if (failoverProxyProvider == null) {
      // Non-HA case
      return createNonHAProxy(conf, NameNode.getAddress(nameNodeUri), xface,
          UserGroupInformation.getCurrentUser(), true, fallbackToSimpleAuth);
    } 

createNonHAProxy执行于

T proxy;
    if (xface == ClientProtocol.class) {
      proxy = (T) createNNProxyWithClientProtocol(nnAddr, conf, ugi,
          withRetries, fallbackToSimpleAuth);
    }

createNNProxyWithClientProtocol 关键代码如下

private static ClientProtocol createNNProxyWithClientProtocol
{
...
ClientNamenodeProtocolPB proxy = RPC.getProtocolProxy(
        ClientNamenodeProtocolPB.class, version, address, ugi, conf,
        NetUtils.getDefaultSocketFactory(conf),
        org.apache.hadoop.ipc.Client.getTimeout(conf), defaultPolicy,
        fallbackToSimpleAuth).getProxy();

...
return new ClientNamenodeProtocolTranslatorPB(proxy);
...

}

由此可以看出 DFSClient namenode 指向的是ClientNamenodeProtocolTranslatorPB 对象

重点 看下 RPC.getProtocolProxy,方法中的关键代码

return getProtocolEngine(protocol, conf).getProxy(protocol, clientVersion,
        addr, ticket, conf, factory, rpcTimeout, connectionRetryPolicy,
        fallbackToSimpleAuth);

getProtocolEngine(protocol, conf) 获得是 ProtobufRpcEngine

ProtobufRpcEngine.getProxy 如下

final Invoker invoker = new Invoker(protocol, addr, ticket, conf, factory,
        rpcTimeout, connectionRetryPolicy, fallbackToSimpleAuth);
    return new ProtocolProxy(protocol, (T) Proxy.newProxyInstance(
        protocol.getClassLoader(), new Class[]{protocol}, invoker), false);

返回的是 ClientNamenodeProtocolPB的代理对象，这个对象 封装在 ProtocolProxy对象中并返回。

从此 DFSClient方法调用 都转换为 ClientNamenodeProtocolPB代理对象的方法调用

看看这个代理究竟做了什么事

这边使用的java动态代理机制，

关键逻辑 在Invoker 的覆写方法中 invoke() 中