Hadoop3.2.1 RPC通讯一锅端

﹏ヽ暗。殇╰゛Y 2023-02-20 02:17 38阅读 0赞

下面是最近整理的关于Hadoop3.2.1  RPC 的通讯文章汇总,

以及我最后画的一张从Client端请求到Server端的流程图. 有什么不对的地方,欢迎指正...

--------------------

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC实现 \[一\] proto接口协议][Hadoop3.2.1 _ HDFS _ _ RPC_ _ proto]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC实现 \[二\] 概述&使用][Hadoop3.2.1 _ HDFS _ _ RPC_ _]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[三\] 概述&原理][Hadoop3.2.1 _ HDFS _ _ RPC_ _ 1]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[四\] RpcEngine父类][Hadoop3.2.1 _ HDFS _ _ RPC_ _ RpcEngine]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[五\] WritableRpcEngine 使用][Hadoop3.2.1 _ HDFS _ _ RPC_ _ WritableRpcEngine]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[六\] ProtobufRpcEngine 使用][Hadoop3.2.1 _ HDFS _ _ RPC_ _ ProtobufRpcEngine]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[七\] Server端实现&源码][Hadoop3.2.1 _ HDFS _ _ RPC_ _ Server]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[八\] Client端实现&源码][Hadoop3.2.1 _ HDFS _ _ RPC_ _ Client]

[Hadoop3.2.1 【 HDFS 】源码分析 : RPC原理 \[九\] RPC 请求格式][Hadoop3.2.1 _ HDFS _ _ RPC_ _ RPC]

--------------------

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70][]

尽管这里是hadoop的rpc服务，但是hadoop还是做到了一次连接仅有一次认证。具体的流程待我慢慢道来。

**客户端**：这里我们假设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。

至此，hdfs中客户端与服务端的交互就比较详细的展现给大家了。

图是我画的,但是懒着打字,直接复制的别人写好的流程.

文字流程来源:

[https://www.cnblogs.com/letsfly/p/10556363.html][https_www.cnblogs.com_letsfly_p_10556363.html]

[Hadoop3.2.1 _ HDFS _ _ RPC_ _ proto]: https://zhangboyi.blog.csdn.net/article/details/105586527
[Hadoop3.2.1 _ HDFS _ _ RPC_ _]: https://zhangboyi.blog.csdn.net/article/details/105591336
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ 1]: https://zhangboyi.blog.csdn.net/article/details/105607870
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ RpcEngine]: https://zhangboyi.blog.csdn.net/article/details/105608092
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ WritableRpcEngine]: https://zhangboyi.blog.csdn.net/article/details/105629868
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ ProtobufRpcEngine]: https://zhangboyi.blog.csdn.net/article/details/105630023
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ Server]: https://zhangboyi.blog.csdn.net/article/details/105634160
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ Client]: https://zhangboyi.blog.csdn.net/article/details/105669375
[Hadoop3.2.1 _ HDFS _ _ RPC_ _ RPC]: https://zhangboyi.blog.csdn.net/article/details/105769434
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200426115046576.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[https_www.cnblogs.com_letsfly_p_10556363.html]: https://www.cnblogs.com/letsfly/p/10556363.html