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

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

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

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

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

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

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

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

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

另外&＃xff0c;如果客户端在请求超时之后&＃xff0c;会调用sendPing方法&＃xff0c;用于测试服务端的服务是否仍然开启&＃xff0c;此时的callId为PING_CALL_ID。

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

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

• 应用信息

资源预约

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

• 集群/节点/配置信息

• 队列信息

Admin与RM之间的通信协议&＃xff0c; Admin通过该RPC协议更新系统配置文件&＃xff0c; 例如节点黑白名单等。ResourceManagerAdministrationProtocol继成了GetUserMappingsProtocol协议接口.
接口信息如下:

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

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

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

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