Hadoop3.2.1 【 YARN 】源码分析 : RPC通讯解析

左手的ㄟ右手 2022-12-26 15:29 205阅读 0赞

### \-------------------------- ###

*  一.前言
 *  二.RPC流程 概略图
 *  三.YARN RPC的主要组成协议
 *  四.ApplicationClientProtocol
 *  五.ResourceManagerAdministrationProtocol
 *  六.ResourceTracker
 *  七.ApplicationMasterProtocol
 *  八.ContainerManagementProtocol

# 一.前言 #

远程过程调用（Remote Procedure Call， RPC） 是一个计算机通信协议。 该协议允许运行于一台计算机的程序调用另一台计算机的子程序， 同时将网络的通信细节隐藏起来，而程序员无需额外地为这个交互作用编程。

作为一个分布式系统， Hadoop实现了自己的RPC通信协议， 它是上层多个分布式子系统（MR、 YARN、 HDFS等） 的公用网络通信模块， 保证其轻量级、 高性能。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

典型的RPC框架， 主要包括以下几个部分。  
（ 1） 通信模块： 主要是请求-应答协议， 包括同步方式和异步方式。  
①Stub程序： 客户端和服务端均包含Stub程序， 可将之看作代理程序。  
②调度程序： 调度程序接收来自通信模块的请求消息， 并根据其中的标识选择一个Stub程序进行处理。  
（ 2） 客户程序/服务过程： 请求的发出者和请求的处理者。  
请求过程为： 客户端程序—>客户端Stub程序—>通信模块—>远程请求—>通信模块—调度程序—服务器Stub程序—>服务程序

# 二.RPC流程 概略图 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

客户端：这里我们假设ConnectionId对应的Connection并不存在。在调用getConnection方法时，这里构造了Connection，由于入参ConnectionId.doPing一般为true，因此，在Connection的构造方法时，会构造相应的pingHeader写入到成员变量pingRequest中。接着将call加入到connection中后，调用了connection.setupIOstreams，这里一开始就调用了writeConnectionHeader，一共写了7个字节的内容到服务端(分别是"hrpc"，Version、Service Class、AuthProtocol，显然，前面是四个字节，后面三个都是一个字节)。另外，由于这里的成员变量doPing为true，因此，这里使用PingInputStream封装了上面获取的输入流。该方法内接着调用了writeConnectionContext，该方法就是rpc的认证调用。这里的callId设置为CONNECTION\_CONTEXT\_CALL\_ID，这里最后调用了request.write，将request中的信息写到了out中，注意，这里并没有调用其flush方法。但是，这里封装了一次请求。接着，调用了connection.sendRpcRequest，这里将请求的call继续写入到out中，并调用了flush方法。至此，客户端的流程也就完成了。

服务端：关于服务端的流程，我从Server.Listener.run讲起。这里一开始是阻塞的。当客户端调用了flush之后，这里的阻塞被打断。这里然后调用到了doAccept方法。在该方法中首先调用到了ConnectionManager.register，这里新构造了Connection并且将其加入到成员结合connections中。在Connection的构造中封装了客户端的channel、socket等相关信息。然后将新构造的connection作为attachment绑定到SelectionKey上，然后调用reader.addConnection将其加入到阻塞队列pendingConnections，并调用wakeup唤醒了reader在doRunLoop方法中的readSelector.select阻塞。而Server.Listener中的方法继续阻塞。

接下来，我们重点关注Server.Listener.Reader.doRunLoop。这里的readSelector.select阻塞打断后，由于此时的readSelector中并没有相关的selectedKeys，因此继续循环执行。接着从pendingConnections队列中取出封装了客户端相关信息的Connection，然后将Connection.channel注册到readSelector选择器上，由于客户端已经将信息flush过来，所以，这里的注册后的readSelector不会阻塞，且其中含有相应的selectedKeys，故，接下来会调用到doRead方法，接着调用到了Connection.readAndProcess。接下来另起一段着重讲解。

这里dataLengthBuffer.remaining()>0判断成立，调用channelRead将客户端的版本信息读入。接着将Version、Service Class、AuthProtocol读入到connectionHeaderBuf中，然后调用了dataLengthBuffer.flip，方便下次读取数据长度，并且将connectionHeaderRead置为true，也就是下一次遍历的时候不会再读取头部的相关信息。只是读取实际数据长度。然后，由于data == null，因此，这里为data分配空间，然后调用了processOneRpc，由于这里是封装的CONNECTION\_CONTEXT\_CALL\_ID的call，因此，这里来到processRpcOutOfBandRequest方法，该方法完成了认证的相关流程。由于一开始connectionContextRead为false，而在processRpcOutOfBandRequest方法中被置为true，因此，这里会调用continue。这里重新读取的数据的长度，然后继续调用了processOneRpc。这一次callId不小于0，因此会调用到方法processRpcRequest。该方法是完成真正请求的，也就是说，前面的几次调用只是前期的校验。所谓真正的请求，也就是要调用服务端的相关方法。processRpcRequest方法将客户端的相关信息封装到Call，并将其加入到callQueue中。

另一方面，在Handler.run中一直在等待callQueue中成员的加入。通过其take方法可以看到内部的阻塞队列一直在轮询等待成员的加入。这里接着调用了CurCall.set(call)，方便后面获取客户端的ugi。接着就调用到了RPC.Server.call，由于这里的Server继承自org.apache.hadoop.ipc.Server(该类是一个抽象类)，而前面的Server覆写了后面Server的抽象方法call。最终调用了ProtobufRpcEngine.cal,该方法完成了服务端对应方法的调用，并且将结果返回。

另外，如果客户端在请求超时之后，会调用sendPing方法，用于测试服务端的服务是否仍然开启，此时的callId为PING\_CALL\_ID。

# 三.YARN RPC的主要组成协议 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

（1） ApplicationClientProtocol： clients与RM之间的协议， JobClient通过该RPC协议提交应用程序、 查询应用程序状态等。  
（2） ResourceManagerAdministrationProtocol： Admin与RM之间的通信协议， Admin通过该RPC协议更新系统配置文件， 例如节点黑白名单等。  
（3） ApplicationMasterProtocol： AM与RM之间的协议， AM通过该RPC协议向RM注册和撤销自己， 并为各个任务申请资源。  
（4） ContainerManagementProtocol： AM与NM之间的协议， AM通过该RPC要求NM启动或者停止Container， 获取各个Container的使用状态等信息。  
（5） ResourceTracker： NM与RM之间的协议， NM通过该RPC协议向RM注册， 并定时发送心跳信息汇报当前节点的资源使用情况和Container运行情况。

# 四.ApplicationClientProtocol #

clients与RM之间的协议， JobClient通过该RPC协议提交应用程序、 查询应用程序状态、 集群状态、 节点、队列和权限控制等。

*  应用信息

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">getNewApplication</td> 
   <td align="left">client获得一个单调递增的ApplicationId用来提交Application .<br>响应信息中会包含集群的一些详细信息,比如集群中指定最大资源能力</td> 
  </tr> 
  <tr> 
   <td align="left">submitApplication</td> 
   <td align="left">client向ResourceManager提交Application. <br>客户端需要通过SubmitApplicationRequest提供诸如队列、运行ApplicationMaster所需的资源、用于启动ApplicationMaster的容器的数据量等详细信息 <br> ResourceManager如果接受请求的话会立即返回一个&lt;空的&gt;SubmitApplicationResponse,如果拒绝则直接回抛出一个异常,然而 请求需要根据getApplicationReport方法确保application被正确的提交, 从ResourceManager获取SubmitApplicationResponse并不保证RM在故障转移或重新启动之后“记住”此应用程序。<br>如果ResourceManager在成功保存application之前发生了故障转义或者重启,getApplicationReport方法将会抛出一个ApplicationNotFoundException异常.当clinet在通过getApplicationReport获得一个ApplicationNotFoundException时,会以相同的ApplicationSubmissionContext配置重新提交application.<br>在提交 的过程中,会检查application 是否存在,如果application存在,将会返回SubmitApplicationResponse.<br>在安全模式下,ResouceManager会在接受应用程序提交之前验证对队列等的访问权限。</td> 
  </tr> 
  <tr> 
   <td align="left">failApplicationAttempt</td> 
   <td align="left">client用来请求ResourceManager使应用程序尝试失败</td> 
  </tr> 
  <tr> 
   <td align="left">forceKillApplication</td> 
   <td align="left">client请求ResourceManager终止已提交的Application</td> 
  </tr> 
  <tr> 
   <td align="left">moveApplicationAcrossQueues</td> 
   <td align="left">将一个application移动到另外一个队列</td> 
  </tr> 
  <tr> 
   <td align="left">updateApplicationPriority</td> 
   <td align="left">更新application的优先级</td> 
  </tr> 
 </tbody> 
</table>

> 资源预约

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">getNewReservation</td> 
   <td align="left">client请求ResourceManager获取ReservationId</td> 
  </tr> 
  <tr> 
   <td align="left">submitReservation</td> 
   <td align="left">提交reservation</td> 
  </tr> 
  <tr> 
   <td align="left">updateReservation</td> 
   <td align="left">更新reservation</td> 
  </tr> 
  <tr> 
   <td align="left">deleteReservation</td> 
   <td align="left">删除reservation</td> 
  </tr> 
  <tr> 
   <td align="left">listReservations</td> 
   <td align="left">根据筛选条件查询reservation列表信息. <br>可通过queue, reservationId, startTime,endTime,includeReservationAllocations等参数.</td> 
  </tr> 
  <tr> 
   <td align="left">signalToContainer</td> 
   <td align="left">客户端用于请求ResourceManager向容器发出信号的接口<br>比如像container发送OUTPUT_THREAD_DUMP命令获取线程快照信息.<br>安全模式下,ResourceManager在application发送命令给container之前会验证权限信息. 用户需要具有&lt;MODIFY_APP&gt;权限</td> 
  </tr> 
  <tr> 
   <td align="left">updateApplicationTimeouts</td> 
   <td align="left">更新application的超时时间. 需要提供&lt;yyyy-MM-dd’T’HH:mm:ss.SSSZ&gt;格式时间,如果超时时间小于等于当前时间会抛出异常</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

1.  用户提交预订(reservation)创建请求，并接收包含ReservationId的响应。
2.  用户提交一个用RDL(Reservation Definition Language)表示的规格文件和ReservationId组成的预订请求。这描述了用户对资源随时间的需求（例如，资源的阈值）和时间约束（例如，期限）。这可以通过常用的Client-to-RM协议或通过RM的REST api以编程方式完成。如果使用相同的ReservationId提交预订，并且RDL相同，则不会创建新预留并且请求将成功。如果RDL不同，则将拒绝预留，并且请求将不成功。如果使用相同的ReservationId提交预订，并且RDL相同，则不会创建新预留但是请求将会成功。如果RDL不同，则将拒绝预留，并且请求将会不成功。
3.  ReservationSystem利用ReservationAgent（图中的GREE）启动一个客户端去查找合理的资源分配。一般是通过一个计划，计划是一个数据结构，描述了当前已经接受的保留和可获得的资源。
4.  SharingPolicy提供一种方式去接受或者拒绝预订。
5.  成功验证后，ReservationSystem会向用户返回ReservationId。  
    6.到一定时间，一个名为PlanFollower的新组件通过动态的创建/调整/销毁队列，将计划状态发布到调度程序。
6.  用户可以提交一个作业到可预留队列，可以明确指定ReservationId作为ApplicationSubmissionContext的一部分。
7.  Scheduler将从创建的特殊队列中提供容器，以确保遵守资源预留。在预留的限制内，用户已经保证访问资源，在资源共享之上进行标准的Capacity/Fairness共享。
8.  系统同时提供一个机制去适应集群容量下降的机制。

*  集群/节点/配置信息

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">getClusterMetrics</td> 
   <td align="left">client 向ResourceManager获取集群信息</td> 
  </tr> 
  <tr> 
   <td align="left">getClusterNodes</td> 
   <td align="left">client 向ResourceManager获取集群所有节点的信息</td> 
  </tr> 
  <tr> 
   <td align="left">getNodeToLabels</td> 
   <td align="left">获取node节点的label信息</td> 
  </tr> 
  <tr> 
   <td align="left">getLabelsToNodes</td> 
   <td align="left">获取label的node 信息</td> 
  </tr> 
  <tr> 
   <td align="left">getClusterNodeLabels</td> 
   <td align="left">获取集群中的label信息</td> 
  </tr> 
  <tr> 
   <td align="left">getResourceProfiles</td> 
   <td align="left">获取特定资源配置[resource-profiles.json,]</td> 
  </tr> 
  <tr> 
   <td align="left">getResourceTypeInfo</td> 
   <td align="left">获取特定资源配置</td> 
  </tr> 
  <tr> 
   <td align="left">getAttributesToNodes</td> 
   <td align="left">获取attribute-&gt;node属性</td> 
  </tr> 
  <tr> 
   <td align="left">getClusterNodeAttributes</td> 
   <td align="left">获取集群node属性</td> 
  </tr> 
  <tr> 
   <td align="left">getNodesToAttributes</td> 
   <td align="left">获取集群node-&gt;attribute属性</td> 
  </tr> 
 </tbody> 
</table>

*  队列信息

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">getQueueInfo</td> 
   <td align="left">获取集群队列信息[已使用/总容量/子队列/正在运行任务信息]</td> 
  </tr> 
  <tr> 
   <td align="left">getQueueUserAcls</td> 
   <td align="left">Client从ResourceManager获取有关当前用户的队列ACL的信息</td> 
  </tr> 
 </tbody> 
</table>

# 五.ResourceManagerAdministrationProtocol #

Admin与RM之间的通信协议， Admin通过该RPC协议更新系统配置文件， 例如节点黑白名单等。ResourceManagerAdministrationProtocol继成了GetUserMappingsProtocol协议接口.  
接口信息如下:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">refreshQueues</td> 
   <td align="left">刷新队列</td> 
  </tr> 
  <tr> 
   <td align="left">refreshNodes</td> 
   <td align="left">刷新节点</td> 
  </tr> 
  <tr> 
   <td align="left">refreshSuperUserGroupsConfiguration</td> 
   <td align="left">刷新配置</td> 
  </tr> 
  <tr> 
   <td align="left">refreshUserToGroupsMappings</td> 
   <td align="left">刷新用户-&gt;用户组映射信息</td> 
  </tr> 
  <tr> 
   <td align="left">refreshAdminAcls</td> 
   <td align="left">刷新Admin的ACL信息</td> 
  </tr> 
  <tr> 
   <td align="left">refreshServiceAcls</td> 
   <td align="left">刷新服务级别信息（SLA）</td> 
  </tr> 
  <tr> 
   <td align="left">updateNodeResource</td> 
   <td align="left">更新在RM端维护的RMNode资源信息</td> 
  </tr> 
  <tr> 
   <td align="left">refreshNodesResources</td> 
   <td align="left">刷新node资源信息</td> 
  </tr> 
  <tr> 
   <td align="left">addToClusterNodeLabels</td> 
   <td align="left">向集群中节点添加Label</td> 
  </tr> 
  <tr> 
   <td align="left">removeFromClusterNodeLabels</td> 
   <td align="left">移除集群中节点Label</td> 
  </tr> 
  <tr> 
   <td align="left">replaceLabelsOnNode</td> 
   <td align="left">替换集群中节点Label</td> 
  </tr> 
  <tr> 
   <td align="left">checkForDecommissioningNodes</td> 
   <td align="left">检查停用的节点</td> 
  </tr> 
  <tr> 
   <td align="left">refreshClusterMaxPriority</td> 
   <td align="left">刷新群集最大优先级</td> 
  </tr> 
  <tr> 
   <td align="left">mapAttributesToNodes</td> 
   <td align="left">获取attribute -&gt; node 信息</td> 
  </tr> 
 </tbody> 
</table>

# 六.ResourceTracker #

NM与RM之间的协议， NM通过该RPC协议向RM注册， 并定时发送心跳信息汇报当前节点的资源使用情况和Container运行情况 \[ 服务端ResourceTrackerService : 8031端口\]。

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">nodeHeartbeat</td> 
   <td align="left">心跳信息</td> 
  </tr> 
  <tr> 
   <td align="left">registerNodeManager</td> 
   <td align="left">注册NodeManager</td> 
  </tr> 
  <tr> 
   <td align="left">unRegisterNodeManager</td> 
   <td align="left">解除注册NodeManager</td> 
  </tr> 
 </tbody> 
</table>

# 七.ApplicationMasterProtocol #

AM与RM之间的协议， AM通过该RPC协议向RM注册和撤销自己， 并为各个任务申请资源。

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">registerApplicationMaster</td> 
   <td align="left">新的ApplicationMaster向RM注册 <br> ApplicationMaster会提供RPC端口,url等信息给RM,响应信息会返回集群所能响应的最大资源能力</td> 
  </tr> 
  <tr> 
   <td align="left">finishApplicationMaster</td> 
   <td align="left">ApplicationMaster通知RM自己的状态为成功/失败</td> 
  </tr> 
  <tr> 
   <td align="left">allocate</td> 
   <td align="left">ApplicationMaster向ResourceManager申请资源/心跳</td> 
  </tr> 
 </tbody> 
</table>

# 八.ContainerManagementProtocol #

ContainerManagementProtocol： AM与NM之间的协议， AM通过该RPC要求NM启动或者停止Container， 获取各个Container的使用状态等信息。

<table> 
 <thead> 
  <tr> 
   <th align="left">方法名称</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">startContainers</td> 
   <td align="left">启动容器</td> 
  </tr> 
  <tr> 
   <td align="left">stopContainers</td> 
   <td align="left">停止容器</td> 
  </tr> 
  <tr> 
   <td align="left">getContainerStatuses</td> 
   <td align="left">获取容器状态</td> 
  </tr> 
  <tr> 
   <td align="left">increaseContainersResource [废弃]</td> 
   <td align="left">增加容器资源</td> 
  </tr> 
  <tr> 
   <td align="left">updateContainer</td> 
   <td align="left">更新容器</td> 
  </tr> 
  <tr> 
   <td align="left">signalToContainer</td> 
   <td align="left">发送信号</td> 
  </tr> 
  <tr> 
   <td align="left">localize</td> 
   <td align="left">本地化容器所需的资源,目前，此API仅适用于运行容器</td> 
  </tr> 
  <tr> 
   <td align="left">reInitializeContainer</td> 
   <td align="left">使用新的 Launch Context 初始化容器</td> 
  </tr> 
  <tr> 
   <td align="left">restartContainer</td> 
   <td align="left">重新启动容器</td> 
  </tr> 
  <tr> 
   <td align="left">rollbackLastReInitialization</td> 
   <td align="left">尝试回滚最后一次重新初始化操作</td> 
  </tr> 
  <tr> 
   <td align="left">commitLastReInitialization</td> 
   <td align="left">尝试提交最后一次初始化操作,如果提交成功则不可以回滚.</td> 
  </tr> 
 </tbody> 
</table>

ContainerManagementProtocol协议主要提供了以下三个RPC函数 :  
❑ startContainer： ApplicationMaster通过该RPC要求NodeManager启动一个Container。该函数有一个StartContainerRequest类型的参数， 封装了Container启动所需的本地资源、 环境变量、 执行命令、 Token等信息。 如果Container启动成功， 则该函数返回一个StartContainerResponse对象。  
❑ stopContainer： ApplicationMaster通过该RPC要求NodeManager停止（ 杀死） 一个Container。 该函数有一个StopContainerRequest类型的参数， 用于指定待杀死的ContainerID。 如果Container被成功杀死， 则该函数返回一个StopContainer-Response对象。  
❑ getContainerStatus： ApplicationMaster通过该RPC获取一个Container的运行状态。 该函数参数类型为GetContainerStatusRequest， 封装了目标Container的ID， 返回值为封装了Container当前运行状态的类型为GetContainerStatusResponse的对象。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221120/dafd25d7c3544f4c8b7c3fa3223e3802.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221120/24f3cd22cfc540a59c391f2d24de70fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221120/508e88c073024110ab227fbda92491ca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221120/415913058c1b4cd9b019857d4153a2f3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/20201207005153893.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70