Hadoop 基于protobuf 的RPC的客户端实现原理

基于google protobuf的RPC engine，必须在服务器端和客户端都完成了初始化之后，才能开始通信。在《Hadoop 基于protobuf 的RPC的服务器端实现原理》这篇博文中，我介绍了RPC 的服务器端实现，那么，客户端是如何基于预先定义的protobuf协议，来与远程的基于相同的protobuf协议的服务端进行通信的呢？

比如，NodeManger与远程的ResourceManager进行RPC通信，它们的通信基于ResourceTracker这个RPC协议，协议定义在ResourceTracker.proto文件中：

service ResourceTrackerService {
rpc registerNodeManager(RegisterNodeManagerRequestProto) returns (RegisterNodeManagerResponseProto);
rpc nodeHeartbeat(NodeHeartbeatRequestProto) returns (NodeHeartbeatResponseProto);
}

协议中定义了两个通信接口，

registerNodeManager

和

nodeHeartbeat

。

registerNodeManager

负责在节点启的NodeManager启动的时候向ResourceManger注册自己，而

nodeHeartbeat

是通过定时心跳的方式，不断向ResourceManager报告自己的存在，并将自己的状态汇报给ResourceManager。

在Yarn的基于Master/Slave的设计模式中，

register

思想是最核心的 设计思想。NodeManager被ResourceManager管理，那么NodeManager在启动的时候必须向ResourceManager注册，RPCEngine想要生效，Engine在初始化的时候也必须向ResourceManager注册。我认为这种设计思想的根本目的，是将主动权交给用户(Slave)而不是管理者(Master)，这样，将Master从繁杂的管理工作中解脱出来，Master不需要关心、不需要轮训NodeManager什么时候来，什么时候启动，而是让NodeManager在启动的时候主动告知即可。

那么，这个客户端协议是怎么进行初始化并向远程的ResourceTracker发送消息的呢？既然NodeManager是该协议的客户端，我们从NodeManager代码进入，来看看初始化以及初始化以后基于协议进行通信的客户端过程。

Yarn代码设计的另外一个重要特点就是功能服务化，无论是NodeManager、ResourceManager还是MapReduce的ApplicationMaster（MRAppMaster）,都抽象为服务，服务之间功能独立，服务的运行 被抽象为初始化、启动、运行和停止等基本过程，让整个代码逻辑非常清晰、封装性变得非常好。

在NodeManager的

serviceInit()

方法中，我们看到：

nodeStatusUpdater =
createNodeStatusUpdater(context, dispatcher, nodeHealthChecker);

protected NodeStatusUpdater createNodeStatusUpdater(Context context,
Dispatcher dispatcher, NodeHealthCheckerService healthChecker) {
return new NodeStatusUpdaterImpl(context, dispatcher, healthChecker,
metrics);
}

创建了一个运行时类型为

NodeStatusUpdaterImpl

的状态更新器，其实，这个NodeStatusUpdater也是一个

service

，启动ResourceTracker协议的客户端，就是在

NodeStatusUpdaterImpl.serviceStart()

中进行：

@Override
protected void serviceStart() throws Exception {

// NodeManager is the last service to start, so NodeId is available.
//...
try {
// Registration has to be in start so that ContainerManager can get the
// perNM tokens needed to authenticate ContainerTokens.
//创建基于ResourceManager协议的RPC客户端
this.resourceTracker = getRMClient();
registerWithRM();
//...
} catch (Exception e) {
//...
}
}

跟踪代码，到

ServerRMProxy.createRMProxy()

:

/**
* Create a proxy for the specified protocol. For non-HA,
* this is a direct connection to the ResourceManager address. When HA is
* enabled, the proxy handles the failover between the ResourceManagers as
* well.
* 对于ResourceTracker协议来说，这里的参数protocol就是是ResourceTracker.class， instanse参数是ServerRMProxy
*/
@Private
protected static <T> T createRMProxy(final Configuration configuration,
final Class<T> protocol, RMProxy instance) throws IOException {
YarnConfiguration conf = (configuration instanceof YarnConfiguration)
? (YarnConfiguration) configuration
: new YarnConfiguration(configuration);
RetryPolicy retryPolicy = createRetryPolicy(conf);
//通过读取配置文件，判断是否开启High Availability模式
if (HAUtil.isHAEnabled(conf)) {//开启HA模式
RMFailoverProxyProvider<T> provider =
instance.createRMFailoverProxyProvider(conf, protocol);
return (T) RetryProxy.create(protocol, provider, retryPolicy);
} else {//不开启HA模式
InetSocketAddress rmAddress = instance.getRMAddress(conf, protocol);
LOG.info("Connecting to ResourceManager at " + rmAddress);
T proxy = RMProxy.<T>getProxy(conf, protocol, rmAddress);
return (T) RetryProxy.create(protocol, proxy, retryPolicy);
}
}

在这里我们开始接触到

proxy

, 如果大家对IPC(Inter-Process Communication，进程间通信)或者RMI(Remote Method Invocation，远程方法调用)不是很熟悉，也许对proxy的理解产生偏差。在这里，proxy指的就是调用者，即客户端。由于在进程间通信或者远程方法调用的时候，调用者只需要调用方法，不需要关心方法是在本地还是远程执行，因此存在一个代理者(即proxy，在java RMI中，也叫做stub程序)，来负责将本地客户端的调用通过TCP等网络协议在远程服务器端进行调用，然后取回调用结果提供给调用者。这就是代理的含义。

是否开启HA模式与本文讨论的话题无关，因此我们选取开启HA模式的分支。继续跟踪代码，看看基于ResourceTracker协议的RPC 客户端是怎么创建的。

跟踪·RMProxy.createRMFailoverProxyProvider()

方法：

/**
* Helper method to create FailoverProxyProvider.
* 同样，
*/
private <T> RMFailoverProxyProvider<T> createRMFailoverProxyProvider(
Configuration conf, Class<T> protocol) {
Class<? extends RMFailoverProxyProvider<T>> defaultProviderClass;
try {
defaultProviderClass = (Class<? extends RMFailoverProxyProvider<T>>)
Class.forName(          YarnConfiguration.DEFAULT_CLIENT_FAILOVER_PROXY_PROVIDER);
} catch (Exception e) {
//some exception
}
//通过配置文件中定义的provider，采用反射方式，建立这个FailoverProxyProvider实例
//默认情况下，这个Provider是
//创建一个Provider，根据默认配置，这个provider是org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider,这个实例的作用是对客户端进行HA的封装代理，似的客户端无需关心HA环境下自己连接的到底是哪个ResourceManager
RMFailoverProxyProvider<T> provider = ReflectionUtils.newInstance(
conf.getClass(YarnConfiguration.CLIENT_FAILOVER_PROXY_PROVIDER,
defaultProviderClass, RMFailoverProxyProvider.class), conf);
provider.init(conf, (RMProxy<T>) this, protocol);
return provider;
}

然后，进入

ConfiguredRMFailoverProxyProvider.init()

方法：

public void init(Configuration configuration, RMProxy<T> rmProxy, Class<T> protocol) {
this.rmProxy = rmProxy;
this.protocol = protocol;
this.rmProxy.checkAllowedProtocols(this.protocol);
//....
}

由此可见，ConfiguredRMFailoverProxyProvider对我们的通信协议进行了HA的封装，在

init

方法中，设置了它所代理的协议名称(ResourceTracker)和这个协议的代理对象RMProxy；在HA环境下，客户端只需要直接使用

ConfiguredRMFailoverProxyProvider

给我们提供的代理对象，而不需要关心这个代理对象到底是指向了哪一个ResourceManager，这就是

ConfiguredRMFailoverProxyProvider

的职责，负责隐藏HA环境下的FailOver细节。

再回到上面提到的代码片段

RMProxy.createRMProxy

：

          // 通过读取配置文件，判断是否开启High Availability模式
if (HAUtil.isHAEnabled(conf)) {// 开启HA模式
//如果开启HA，则provider是一个org.apache.hadoop.yarn.client.ConfiguredRMFailoverProxyProvider
RMFailoverProxyProvider<T> provider = instance.createRMFailoverProxyProvider(conf, protocol);//instance是ServerRMProxy
return (T) RetryProxy.create(protocol, provider, retryPolicy);
} else {// 不开启HA模式
//.....
}

RMFailoverProxyProvider<T> provider = instance.createRMFailoverProxyProvider(conf, protocol);

负责创建和初始化ResourceTracker协议在HA环境下的代理

ConfiguredRMFailoverProxyProvider

，那么，

ConfiguredRMFailoverProxyProvider

是怎么创建真正的RPC客户端的呢？

我们继续跟踪下一行代码

(T) RetryProxy.create(protocol, provider, retryPolicy);

,此时，protocol是ResourceTracker.class，provider是

ConfiguredRMFailoverProxyProvider

:

public static <T> Object create(Class<T> iface,
FailoverProxyProvider<T> proxyProvider, RetryPolicy retryPolicy) {
return Proxy.newProxyInstance(
proxyProvider.getInterface().getClassLoader(),
new Class<?>[] { iface },
new RetryInvocationHandler<T>(proxyProvider, retryPolicy)
);
}

可以看到，ConfiguredRMFailoverProxyProvider通过java动态代理来代理了ResourceTracker协议里面方法的执行。熟悉java动态代理的都会明白，每一个动态代理proxy都需要有继承

java.lang.reflect.InvocationHandler

并实现其

invoke()

方法，用来代替被代理类的执行，这里，这个

InvocationHandler

就是RetryInvocationHandler。ConfiguredRMFailoverProxyProvider底层真正的RPC（已经说过，ConfiguredRMFailoverProxyProvider就是对真正的RPC封装了一层HA特性），就是RetryInvocationHandler来实现的，其实是在RetryInvocationHandler的构造方法里面进行的：

protected RetryInvocationHandler(FailoverProxyProvider<T> proxyProvider,
RetryPolicy defaultPolicy,
Map<String, RetryPolicy> methodNameToPolicyMap) {
//....
this.currentProxy = proxyProvider.getProxy();//proxyProvider其实是ConfiguredRMFailoverProxyProvider对象
}

@Override
public synchronized ProxyInfo<T> getProxy() {
String rmId = rmServiceIds[currentProxyIndex];
T current = proxies.get(rmId);
if (current == null) {
current = getProxyInternal();
proxies.put(rmId, current);
}
return new ProxyInfo<T>(current, rmId);
}

接着往下跟踪：

/**
* 负责创建客户端代理对象
* @return
*/
private T getProxyInternal() {
try {
//通过配置文件，获取远程
//rmProxy的运行时对象是ServerRMProxy，protocol是ResourceTracker.class
final InetSocketAddress rmAddress = rmProxy.getRMAddress(conf, protocol);
return RMProxy.getProxy(conf, protocol, rmAddress);
} catch (IOException ioe) {
LOG.error("Unable to create proxy to the ResourceManager " +
rmServiceIds[currentProxyIndex], ioe);
return null;
}

创建远程ResourceManager服务器端的地址对象，即ResourceTracker协议的服务器端地址信息

@InterfaceAudience.Private
@Override
protected InetSocketAddress getRMAddress(YarnConfiguration conf,
Class<?> protocol) {
if (protocol == ResourceTracker.class) {
//
return conf.getSocketAddr(
YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS,
YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_ADDRESS,
YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_PORT);
//YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS的值是yarn.resourcemanager.resource-tracker.addresss，conf.getSocketAddr会从配置文件中读取YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS对应的配置项，如果没有配置，就使用默认值YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_ADDRESS，默认是0.0.0.0:8031,端口默认值是YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_PORT ，8031
} else {
//Throw some exceptions
}
}

显然，

getRMAddress()

方法就是通过读取配置文件来创建了一个

InetSocketAddress

对象，然后，真正底层创建proxy的时刻到来，来看

RMProxy.getProxy()

：

* Get a proxy to the RM at the specified address. To be used to create a
* RetryProxy.
* 对于ResourceTracker协议来说，这里的参数protocol就是ResourceTracker.class
*/
@Private
static <T> T getProxy(final Configuration conf, final Class<T> protocol, final InetSocketAddress rmAddress)
throws IOException {
return UserGroupInformation.getCurrentUser().doAs(new PrivilegedAction<T>() {
@Override
public T run() {
//Yarn的所有RPC客户端和服务器端都是用YarnRPC进行创建
return (T) YarnRPC.create(conf).getProxy(protocol, rmAddress, conf);
}
});
}

YarnRPC是一个抽象类(Abstract Class)，是Yarn对Hadoop RPC 的封装，基于历史原因和版本升级迭代，Hadoop RPC有基于多种序列化方式的RPC协议，但是由于Yarn是Hadoop 2.0之后才有的组件，是很新的component， 因此Yarn所有的RPC调用都是基于google protobuf序列化方式的RPC进行的实现。

我们一起来看YarnRPC的类图：



YarnRPC.create()

public static YarnRPC create(Configuration conf) {
LOG.debug("Creating YarnRPC for " +
conf.get(YarnConfiguration.IPC_RPC_IMPL));
//yarn.ipc.rpc.class  默认是org.apache.hadoop.yarn.ipc.HadoopYarnProtoRPC
String clazzName = conf.get(YarnConfiguration.IPC_RPC_IMPL);
if (clazzName == null) {
clazzName = YarnConfiguration.DEFAULT_IPC_RPC_IMPL;
}
try {
return (YarnRPC) Class.forName(clazzName).newInstance();
} catch (Exception e) {
throw new YarnRuntimeException(e);
}
}

YarnRPC这个抽象类的实际实现类的名称是通过Yarn配置文件读取，默认是

org.apache.hadoop.yarn.ipc.HadoopYarnProtoRPC

这个类，

因此

YarnRPC.create(conf).getProxy(protocol, rmAddress, conf);

实际上调用了

HadoopYarnProtoRPC.getProxy()

方法:

public Object getProxy(Class protocol, InetSocketAddress addr, Configuration conf) {
LOG.debug("Creating a HadoopYarnProtoRpc proxy for protocol " + protocol);
// 默认的clientfactory是org.apache.hadoop.yarn.factories.impl.pb.RpcClientFactoryPBImpl
return RpcFactoryProvider.getClientFactory(conf).getClient(protocol, 1, addr, conf);
}

进入

RpcClientFactoryPBImpl.getClient()

:

//对于ResourceTracker协议来说，这里的参数protocol就是ResourceTracker.class
public Object getClient(Class<?> protocol, long clientVersion,
InetSocketAddress addr, Configuration conf) {
Constructor<?> constructor = cache.get(protocol);
if (constructor == null) {
Class<?> pbClazz = null;
try {
//根据Yarn自身的规定，需要根据protocol名称拿到具体的客户端实现，即从ResourceTracker -> ResourceTrackerPBClientImpl，所有的Yarn RPC都遵循这样的转换规定，除了ResourceTracker，还有比如ApplicationMaster和ResourceManager进行沟通的协议ApplicationMasterProtocol，它对应的客户端实现类叫做ResourceTrackerPBClientImpl，而对应的服务器端实现类叫做ResourceTrackerPBServerImpl
pbClazz = localConf.getClassByName(getPBImplClassName(protocol));
} catch (ClassNotFoundException e) {
//some exceptions
}
try {
//ResourceTrackerPBClientImpl的构造函数
constructor = pbClazz.getConstructor(Long.TYPE, InetSocketAddress.class, Configuration.class);
constructor.setAccessible(true);
cache.putIfAbsent(protocol, constructor);
} catch (NoSuchMethodException e) {
//some exceptions
}
}
try {
//构造ResourceTrackerPBClientImpl对象
Object retObject = constructor.newInstance(clientVersion, addr, conf);
return retObject;
} catch (InvocationTargetException e) {
///some exceptions
}
}

ResourceTrackerPBClientImpl就是对ResourceTracker协议的最下层代理，来看ResourceTrackerPBClientImpl的构造函数：

public ResourceTrackerPBClientImpl(long clientVersion, InetSocketAddress addr, Configuration conf)
throws IOException {
//为protocol注册处理引擎
RPC.setProtocolEngine(conf, ResourceTrackerPB.class, ProtobufRpcEngine.class);
//设置并获取protocol的代理类
proxy = (ResourceTrackerPB) RPC.getProxy(ResourceTrackerPB.class, clientVersion, addr, conf);
}

在这里，我们再次看到了hadoop中的注册思想。我们的ResourceTrackerPBClientImpl协议要想使用，必须向对应的RPC Engine注册自己。所有基于protobuf协议的RPC都必须向ProtobufRpcEngine进行注册，注册完成以后，创建底层的客户端代理。

终于，经过繁杂但是设计良好的Protobuf RPC的初始化，我们终于拿到了ResourceTracker协议的客户端实现类。此后，ResourceTracker协议的客户端，我们的 NodeManager，就可以根据协议的定义，来进行协议中的

registerNodeManager()

方法的调用。我们跟踪一下这个过程，试图搞清楚客户端在调用这个方法的时候，是如何不知不觉通过RPC变成了服务器端的调用的。

registerNodeManager()

是由NodeManager发起的，NodeManager实际上是委托

NodeStatusUpdaterImpl

来与服务器端的ResourceManager进行沟通，看

NodeStatusUpdaterImpl.registerNodeManager

:

@Override
public RegisterNodeManagerResponse registerNodeManager(
RegisterNodeManagerRequest request) throws YarnException,
IOException {
RegisterNodeManagerRequestProto requestProto = ((RegisterNodeManagerRequestPBImpl)request).getProto();
try {

return new RegisterNodeManagerResponsePBImpl(proxy.registerNodeManager(null, requestProto));
} catch (ServiceException e) {
RPCUtil.unwrapAndThrowException(e);
return null;
}
}

这个proxy对象，是 ResourceTrackerPBClientImpl构造函数执行的时候创建的：

由于Yarn RPC使用Protobuf ，因此

RPC.getProxy

实际上调用的是

ProtobufRpcEngine().getProxy()

方法：

@Override
@SuppressWarnings("unchecked")
public <T> ProtocolProxy<T> getProxy(Class<T> protocol, long clientVersion,
InetSocketAddress addr, UserGroupInformation ticket, Configuration conf,
SocketFactory factory, int rpcTimeout, RetryPolicy connectionRetryPolicy,
AtomicBoolean fallbackToSimpleAuth) throws IOException {

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

很明显，这里是通过java动态代理，来对

ResourceTrackerPBClientImpl

的方法进行代理执行。再次重复上面关于java 动态代理的解释，所有java动态代理都必须实现

java.lang.reflect.InvocationHandler

接口，实现其

invoke()

方法，用来代替被代理类的执行，对于

ProtobufRpcEngine

，这个

InvocationHandler

就是

ProtobufRpcEngine.Invoker

。我们看

ProtobufRpcEngine.Invoker

是怎么代理这个客户端的registerNodeManager()方法的执行的：

@Override
public Object invoke(Object proxy, Method method, Object[] args)
throws ServiceException {
long startTime = 0;
//some code
//开始将方法的信息和请求信息进行包装，准备发送给server
RequestHeaderProto rpcRequestHeader = constructRpcRequestHeader(method);
//....
//提取请求方法的参数
Message theRequest = (Message) args[1];
final RpcResponseWrapper val;
try {
//将请求信息发送给远程服务器
//remoteId是一个 Client.ConnectionId，封装了该协议对应的远程服务器的信息，比如ip、端口等
//RpcRequestWrapper封装了请求的方法、参数信息，并且RpcRequestWrapper是一个Writable，因此
//可以被序列化然后发送给远端
val = (RpcResponseWrapper) client.call(RPC.RpcKind.RPC_PROTOCOL_BUFFER,
new RpcRequestWrapper(rpcRequestHeader, theRequest), remoteId,
fallbackToSimpleAuth);

} catch (Throwable e) {
//some exceptions
} finally {
if (traceScope != null) traceScope.close();
}
Message prototype = null;
try {
prototype = getReturnProtoType(method);
} catch (Exception e) {
throw new ServiceException(e);
}
Message returnMessage;
try {
returnMessage = prototype.newBuilderForType()
.mergeFrom(val.theResponseRead).build();

} catch (Throwable e) {
throw new ServiceException(e);
}
return returnMessage;
}

可以看到，ProtobufRpcEngine.Invokder.invoke()方法做的工作，就是提取客户端请求的方法以及方法的参数，将这些信息发送给远程服务器。远程服务器再通过解析，提取出方法和方法参数，在服务器端本地执行对应的代码，比如，服务器端从客户端请求中提取了方法名称为registerNodeManager()以及参数（包含了节点信息等等），会将节点信息进行注册和管理，然后返回注册成功信息。

在HA环境下，通过一层层代理封装，Yarn实现了HA环境下的ResourceManager协议客户端，ResourceTrackerPBClientImpl封装了该协议的客户端实现，属于下层代理，通过这个下层动态代理，将客户端对应方法的调用，转换成字节码信息发送给远端，而

ConfiguredRMFailoverProxyProvider

也是通过动态代理，在ResourceTrackerPBClientImpl的上层进行了封装，以实现High Availability特性。在HA环境下，NodeManager作为ResourceTracker客户端，从ConfiguredRMFailoverProxyProvider的上层代理往下调用，到达ResourceTrackerPBClientImpl下层代理，然后ResourceTrackerPBClientImpl通过动态代理，将请求信息发送到RPC Server，实现了该协议的一次调用。