Hadoop RPC使用与实现

 

hadoop.ipc和RPC简介

hadoop 和hbase中的大部分服务都是通过hadoop.ipt.RPC这个类来实现的。

hadoop.ipc.RPC 实现了一种远程过程调用的框架，应用可以直接定义过程调用的协议接口和协议的server端实现，就可以直接通过RPC框架获得RPC server和client端的接口代理。

hadoop.ipc.RPC 的实现利用了 hadoop.ipc.Server 和 hadoop.ipc.Client这两个类， 这两个类实现了网络中非常典型的Request-Response模式服务器和客户端框架。用户可以通过定义一个协议接口并实现出Request和Response类，以及Server端的抽象处理接口(Server.call()) 就可以实现出完整的服务器程序，而客户端程序只需要在创建hadoop.ipc.Client实体时，指定协议接口和网络相关参数，然后调用 call() 就可以发送请求并获取响应。

Hadoop.ipc.RPC作为Hadoop的底层核心组件，在hadoop HDFS，MapReduce以及HBase中都有广泛的使用。 HDFS中NameNode，DataNode等都是通过实现对应协议的接口，然后利用hadoop.ipc.RPC获取服务器实体的。 HBase中的HBaseRPC采用的也是与hadoop.ipc.RPC类似的实现，其中的Region Server, Master Server 都是通过实现对应的协议接口直接获取服务器实体的。  

hadoop.ipc将应用逻辑与网络消息的处理分离开，并且使得逻辑对象在不同的进程或组件之间有同样的语言接口，无需区分远程对象和本地对象，使得开发者可以关注于应用的处理逻辑。

hadoop.ipc.RPC类中有两个重要的函数getServer和getProxy，getServer通过接口协议实现的实体来获取真正的server，getProxy获取远程访问的本地代理。

 

class RPC {
    public static Server getServer(final Object instance, final String bindAddress, final int port, Configuration conf) 
    public static VersionedProtocol getProxy(
      Class<? extends VersionedProtocol> protocol,
      long clientVersion, InetSocketAddress addr, Configuration conf,
      SocketFactory factory)
    ......
}

 

 

 

实例(远程执行Shell命令)

定义客户端与服务器的协议接口ExecProtocol

import java.lang.String;
import org.apache.hadoop.ipc.VersionedProtocol;
/* 这里不扩展VersionedProtocol 也是可以的 */
public interface ExecProtocol extends VersionedProtocol
{
    public static final long versionID = 1L;
    public String exec(String[] cmd);
}

ExecServer 实现了ExecProtocol，并通过RPC.getServer 获取RPCserver

import java.lang.String;
import java.io.IOException;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.conf.Configuration;
import java.io.BufferedInputStream;
import java.io.InputStream;

public class ExecServer implements ExecProtocol
{
    private String host;
    private int port;
    private RPC.Server server;
    public ExecServer (String host, int port) throws IOException
    {
        this.host = host;
        this.port = port;
        /* 获取server实体 */
        this.server = RPC.getServer (this, host, port, new Configuration ());
    }

    public void run () throws IOException
    {
        /* 运行 */
        this.server.start ();
    }
    /* 实现 VersionedProtocol 接口 */
    public long getProtocolVersion (String s, long v)
    {
        return versionID;
    }
    /* 实现 ExecProtocol.exec接口 */
    public String exec (String[] cmd)
    {
        try {
            Process process = Runtime.getRuntime ().exec (cmd);
            process.waitFor();
            return loadStream (process.getInputStream ())
                   + loadStream (process.getErrorStream ());
        } catch (Exception e) {
            return e.getMessage ();
        }
    }
    /* 用于实现具体的exec接口 */
    private static String loadStream (InputStream stream) throws IOException
    {
        if (stream == null) {
            throw new java.io.IOException ("null stream");
        }
        stream = new java.io.BufferedInputStream (stream);
        int avail = stream.available ();
        byte[]data = new byte[avail];
        int numRead = 0;
        int pos = 0;
        do {
            if (pos + avail > data.length) {
                byte[]newData = new byte[pos + avail];
                System.arraycopy (data, 0, newData, 0, pos);
                data = newData;
            }
            numRead = stream.read (data, pos, avail);
            if (numRead >= 0) {
                pos += numRead;
            }
            avail = stream.available ();
        } while (avail > 0 && numRead >= 0);
        return new String (data, 0, pos, "US-ASCII");
    }

    public static void main (String[]args) throws IOException
    {
        ExecServer s = new ExecServer ("localhost", 1600);
        s.run ();
    }
}

 

client 端通过RPC.getProxy获取本地代理

import java.lang.String;
import java.io.IOException;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.conf.Configuration;
import java.net.InetSocketAddress;
public class ExecClient
{
    public static void main (String[] args) throws IOException
    {
        /* 服务器地址 */
        InetSocketAddress addr = new InetSocketAddress ("localhost", 1600); 
        /* 通过RPC.getProxy 获取客户端代理类的实体 */
        ExecProtocol proxy = (ExecProtocol) RPC.getProxy(ExecProtocol.class,
                                ExecProtocol.versionID, addr,
                                new Configuration ());
        /* 输出 */
        System.out.print (proxy.exec(args));
    }
}

 

Hadoop RPC的实现

 

Hadoop.ipc.RPC的实现依赖于ipc.Server 和ipc.Client。Hadoop.ipc.RPC 将过程调用封装成具体的Invocation类，该类封装了调用的方法和参数，并用一个ObjectWritable类来定义返回的数据类型。ipc.Server 和 ipc.Client 负责网络数据的收发。

 

ipc.Server 的实现包含以下几个部分

Listener 监听网络端口，接受网络请求，然后交给Reader处理

Reader 非阻塞收取网络数据，并解析

Handler 调用抽象方法Server.call生成相应数据

Responder 非阻塞发送数据

Server.call() 由具体RPC.Server利用反射和代理实现，处理请求

这是一种典型的流水线结构，采用流水线的原因是Handler的操作可能导致阻塞，必须要有独立的线程或线程组处理Hander，线程之间的切换必不可少。

Listener独立为单独的线程大概是为了Reader之间负载的均衡，新加入的连接按照round robin在Reader之间进行负载均衡（实际上可能并不均衡，每条连接处理的请求以及持续的时间是不确定的）。同步时需要注意的一点是，Listener在向Reader的Selector中添加链接时，需要设置一个adding标记，并打断Selecter，这样做的目的是避免Reader的select() 操作和register()操作产生竞态。