Hadoop RPC机制学习和使用

RPC，即Remote Procdure Call，中文名：远程过程调用；

RPC的基础概念

• 它允许一台计算机程序远程调用另外一台计算机的子程序，而不用去关心底层的网络通信细节，对我们来说是透明的。因此，它经常用于分布式网络通信中。

• Hadoop的进程间交互都是通过RPC来进行的，比如Namenode与Datanode直接，Jobtracker与Tasktracker之间等。

RPC的显著特点

1. 透明性：远程调用其他机器上的程序，对用户来说就像是调用本地方法一样；
2. 高性能：RPC Server能够并发处理多个来自Client的请求；
3. 可控性：jdk中已经提供了一个RPC框架—RMI，但是该PRC框架过于重量级并且可控之处比较少，所以Hadoop RPC实现了自定义的PRC框架。

RPC的基本流程

（1）RPC采用了C/S的模式；

（2）Client端发送一个带有参数的请求信息到Server；

（3）Server接收到这个请求以后，根据发送过来的参数调用相应的程序，然后把自己计算好的结果发送给Client端；

（4）Client端接收到结果后继续运行；

Hadoop中的RPC机制

同其他RPC框架一样，Hadoop RPC分为四个部分：

（1）序列化层：Clent与Server端通信传递的信息采用了Hadoop里提供的序列化类或自定义的Writable类型；

（2）函数调用层：Hadoop RPC通过动态代理以及java反射实现函数调用；

（3）网络传输层：Hadoop RPC采用了基于TCP/IP的socket机制；

（4）服务器端框架层：RPC Server利用java NIO以及采用了事件驱动的I/O模型，提高RPC Server的并发处理能力；

Hadoop RPC设计技术

• 动态代理

• 反射——动态加载类
• 序列化
• 非阻塞的异步IO（NIO）

Hadoop RPC对外提供的接口

Hadoop RPC对外主要提供了两种接口（见类org.apache.hadoop.ipc.RPC），分别是：

（1）public static <T> ProtocolProxy <T> getProxy/waitForProxy(…)

构造一个客户端代理对象（该对象实现了某个协议），用于向服务器发送RPC请求。

（2）public static Server RPC.Builder (Configuration).build()

为某个协议（实际上是Java接口）实例构造一个服务器对象，用于处理客户端发送的请求。

使用Hadoop RPC的四大步骤

（1）定义RPC协议

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

（2）实现RPC协议

Hadoop RPC协议通常是一个Java接口，用户需要实现该接口。

（3）构造和启动RPC SERVER

直接使用静态类Builder构造一个RPC Server，并调用函数start()启动该Server。

（4）构造RPC Client并发送请求

使用静态方法getProxy构造客户端代理对象，直接通过代理对象调用远程端的方法。

定义RPC协议

如下所示，我们定义一个IProxyProtocol 通信接口，声明了一个Add()方法。

public interface IProxyProtocol extends VersionedProtocol {
    static final long VERSION = 23234L; //版本号，默认情况下，不同版本号的RPC Client和Server之间不能相互通信
    int Add(int number1,int number2);
}

需要注意的是：

（1）Hadoop中所有自定义RPC接口都需要继承VersionedProtocol接口，它描述了协议的版本信息。

（2）默认情况下，不同版本号的RPC Client和Server之间不能相互通信，因此客户端和服务端通过版本号标识。

实现RPC协议

Hadoop RPC协议通常是一个Java接口，用户需要实现该接口。对IProxyProtocol接口进行简单的实现如下所示：

public class MyProxy implements IProxyProtocol {
    public int Add(int number1,int number2) {
        System.out.println("我被调用了!");
        int result = number1+number2;
        return result;
    }

    public long getProtocolVersion(String protocol, long clientVersion)
            throws IOException {
        System.out.println("MyProxy.ProtocolVersion=" + IProxyProtocol.VERSION);
        // 注意：这里返回的版本号与客户端提供的版本号需保持一致
        return IProxyProtocol.VERSION;
    }
}

这里实现的Add方法很简单，就是一个加法操作。为了查看效果，这里通过控制台输出一句：“我被调用了！”

构造RPC Server并启动服务

这里通过RPC的静态方法getServer来获得Server对象，如下代码所示：

public class MyServer {
    public static int PORT = 5432;
    public static String IPAddress = "127.0.0.1";

    public static void main(String[] args) throws Exception {
        MyProxy proxy = new MyProxy();
        final Server server = RPC.getServer(proxy, IPAddress, PORT, new Configuration());
        server.start();
    }
}

这段代码的核心在于第5行的RPC.getServer方法，该方法有四个参数，第一个参数是被调用的java对象，第二个参数是服务器的地址，第三个参数是服务器的端口 。获得服务器对象后，启动服务器。这样，服务器就在指定端口监听客户端的请求。到此为止，服务器就处于监听状态，不停地等待客户端请求到达。

构造RPC Client并发出请求

这里使用静态方法getProxy或waitForProxy构造客户端代理对象，直接通过代理对象调用远程端的方法，具体如下所示：

public class MyClient {

    public static void main(String[] args) {
        InetSocketAddress inetSocketAddress = new InetSocketAddress(
                MyServer.IPAddress, MyServer.PORT);

        try {
            // 注意：这里传入的版本号需要与代理保持一致
            IProxyProtocol proxy = (IProxyProtocol) RPC.waitForProxy(
                    IProxyProtocol.class, IProxyProtocol.VERSION, inetSocketAddress,
                    new Configuration());
            int result = proxy.Add(10, 25);
            System.out.println("10+25=" + result);

            RPC.stopProxy(proxy);
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

以上代码中核心在于RPC.waitForProxy()，该方法有四个参数，第一个参数是被调用的接口类，第二个是客户端版本号，第三个是服务端地址。返回的代理对象，就是服务端对象的代理，内部就是使用java.lang.Proxy实现的。

经过以上四步，我们便利用Hadoop RPC搭建了一个非常高效的客户机–服务器网络模型。

查看运行结果

（1）启动服务端，开始监听客户端请求

（2）启动客户端，开始向服务端发请求

（3）查看服务端状态，是否被调用

（4）现在我们在命令行执行jps命令，查看输出信息，会出现如下图所示的：

从上图中可以看到一个java进程，是“MyServer”，该进程正是我们刚刚运行的RPC的服务端类MyServer。因此，大家可以联想到我们搭建Hadoop环境时，也执行过该命令用来判断Hadoop的相关进程是否全部启动。

下面我们观察NameNode的源代码，如下图所示，可以看到NameNode确实创建了RPC的服务端。

private void initialize(Configuration conf) throws IOException {
    ......
    // create rpc server
    InetSocketAddress dnSocketAddr = getServiceRpcServerAddress(conf);
    if (dnSocketAddr != null) {
      int serviceHandlerCount =
        conf.getInt(DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_KEY,
                    DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_DEFAULT);
      this.serviceRpcServer = RPC.getServer(this, dnSocketAddr.getHostName(), 
          dnSocketAddr.getPort(), serviceHandlerCount,
          false, conf, namesystem.getDelegationTokenSecretManager());
      this.serviceRPCAddress = this.serviceRpcServer.getListenerAddress();
      setRpcServiceServerAddress(conf);
    }
    this.server = RPC.getServer(this, socAddr.getHostName(),
        socAddr.getPort(), handlerCount, false, conf, namesystem
        .getDelegationTokenSecretManager());
   ......
}

博文转载于：http://www.cnblogs.com/edisonchou/p/4285817.html

相关推荐

Hadoop Yarn 框架原理及运作机制

Hadoop自定义输出文件格式和文件输出路径