RPC--HDFS节点间的沟通桥梁

Posted 2021-04-11 数据未来之路

    RPC(Remote Procedure Call, 远程过程调用)是一种通过网络从远程计算机上请求服务来得到计算服务或者数据服务，且不需要了解底层网络技术的协议和框架。

    RPC远程调用是构建在语言级别的，必须使用Socket通信完成，将现有的本地方法调用和Socket网络通信技术结合起来实现透明的远程调用过程。实现透明的远程调用重点是创建客户存根(client stub)，存根(stub)就像代理(agent)模式里的代理(agent)，在生成代理代码后，代理代码与远程服务端通信，通信过程由RPC框架实现，调用者像调用本地代码一样方便。在客户端看来，存根函数像普通本地函数一样，实际包含了网络发送和接收消息的代码。具体过程如下图所示。

1. 客户端调用本地客户端的存根方法(client stub)，将参数打包并封装成一个或多个网络消息体，这个过程被称为编码(encode)，它会将所有数据序列化成字节数组格式;

2. client stub通过系统调用，使用操作系统内核提供的Socket套接字接口来向远程服务发送编码的网络消息；

3. 网络消息由内核通过网络协议(无连接协议：UDP, 或面向连接的协议：TCP)传输到远程服务端；

4. 服务端存根(server stub)接收客户端发送的消息，并对参数消息进行解码(decode)，通常它会将参数从标准的网络格式转换成特定的语言格式；

5. server stub调用服务端方法，并将从客户端接收的参数传递给该方法；

6. 服务端方法执行具体的功能并返回，此部分代码的执行对客户端而言就是远程过程调用；

7. 服务端在方法执行完后，讲结果返回到服务端存根代码中；

8. 服务端存根在降该返回值进行编码并序列化后，通过一个或多个网络消息发送给客户端；

9. 消息通过网络发送到客户端存根；

10. 客户端存根从本地Socket接口中读取结果消息；

11. 客户端存根将结果返回给客户端函数，并将消息从网络二进制形式转换为本地语言格式，这样就完成了远程服务调用，客户端代码继续执行后续操作。

    Socket是客户端/服务端网络结构程序的基本组成部门，为程序提供了一种相对简单的机制与远程机器上的程序建立连接，并来回发送消息。基于这些收发功能，将RPC的调用包装成透明的远程服务调用。

    HDFS的底层通信是通过RPC和动态代理对象Proxy实现的。Hadoop RPC的使用方法分为以下几步：

1. 定义RPC协议。RPC协议是客户端和服务器之间的通信接口，它定义了服务器对外提供的服务接口；

2. 实现RPC协议。Hadoop RPC是一个Java接口，需要用户实现。

3. 构造并启动RPC Server。直接使用静态方法getServer()构造一个RPC Server，并调用start()方法启动该Server。

4. 构造RPC Client，并发送RPC请求。使用静态方法getProxy()构造客户端代理对象，直接通过代理对象调用远程端的方法。

HDFS RPC代码示例：

    服务端代码

package beichen.rpc.server;import org.apache.hadoop.ipc.VersionedProtocol;public interface MyInterface extends VersionedProtocol{    //定义一个版本号，使用版本号来进行签名  public static long versionID = 1;   //定义业务方法 public String sayHello(String name);}
package beichen.rpc.server;import java.io.IOException;import org.apache.hadoop.ipc.ProtocolSignature;public class MyInterfaceImpl implements MyInterface {
 @Override public String sayHello(String name) { System.out.println("*********调用到了Server 端**********"); return "Hello "+name; }
 @Override public ProtocolSignature getProtocolSignature(String arg0, long arg1, int arg2) throws IOException { //通过版本号定义签名信息 return new ProtocolSignature(MyInterface.versionID, null); }
 @Override public long getProtocolVersion(String arg0, long arg1) throws IOException { //返回版本号 return MyInterface.versionID; }}
package beichen.rpc.server;import java.io.IOException;import org.apache.hadoop.HadoopIllegalArgumentException;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.ipc.RPC;import org.apache.hadoop.ipc.RPC.Server;public class MyRPCServer { public static void main(String[] args) throws Exception { //利用Hadoop RPC的框架实现RPC Server,使用RPC Builder来构建  RPC.Builder builder = new RPC.Builder(new Configuration());  //定义Server的参数 builder.setBindAddress("localhost"); builder.setPort(7788);  //部署程序 builder.setProtocol(MyInterface.class); //部署的接口 builder.setInstance(new MyInterfaceImpl()); //指定接口的实现类  // 创建RPC Server Server server = builder.build();  //启动Server server.start(); }}

    客户端代码

package beichen.rpc.client;import java.io.IOException;import java.net.InetSocketAddress;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.ipc.RPC;import beichen.rpc.server.MyInterface;public class MyRPCClient {  // 使用Hadoop RPC的框架调用Server 端的程序  public static void main(String[] args) throws Exception {   //得到的是Server端部署对象的代理对象 MyInterface proxy = RPC.getProxy(MyInterface.class,  MyInterface.versionID,  new InetSocketAddress("localhost", 7788),  new Configuration());  //使用这个代理对象调用Server的程序 String result = proxy.sayHello("Tom"); System.out.println(result); }}