hadoop的RPC分析

      hadoop在实现datanode、namenode、client之间的通信时，实现了自己的一套rpc通信的协议，协议服务器端采用nio的方式来处理请求，支持局域网的rpc调用 。

      协议的传输数据采用writeable数据，每次调用将具体函数参数（writeable），调用方法名称，调用参数类型信息传送过去，然后Server端接收到这些参数之后再根据该方法名称，调用参数类型信息得到相应的Method对象，然后使用参数调用 。

      注释源代码见：https://files.cnblogs.com/serendipity/ipc.rar

     实例代码：

     Client端测试代码：

package com.gemantic.rpc;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.net.NetUtils;
import org.junit.Test;

import com.gemantic.rpc.RPCServer.RPCTestInterface;

public class RPCClientTest {

    /**
     * @param args
     * @throws IOException 
     */
    public static void main(String[] args) throws IOException {

        InetSocketAddress nameNodeAddr = NetUtils.createSocketAddr("127.0.0.1:9813"); 
        RPCTestInterface rpcInterface=(RPCTestInterface)RPC.getProxy(RPCTestInterface.class, RPCTestInterface.versionID, nameNodeAddr, new Configuration());
        rpcInterface.getUper("aedfrtyh");
        System.out.println("--------------------------------------");
        rpcInterface.getUper("dfsgert");
        System.out.println("--------------------------------------");
        rpcInterface.getUper("ertyui");
        System.out.println("--------------------------------------");
        rpcInterface.getUper("qwert");
        
    }
    

}

     Server端测试代码：

package com.gemantic.rpc;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.VersionedProtocol;
import org.apache.hadoop.ipc.RPC.Server;
import org.junit.Test;




public class RPCServer {

    /**
     * @param args
     * @throws IOException 
     * @throws InterruptedException 
     */
    public static void main(String[] args) throws IOException, InterruptedException {
        // TODO Auto-generated method stub

        Server server=RPC.getServer(new RPCTest(), "127.0.0.1", 9813,6, true ,new Configuration());
        server.start();
        server.join();
    }

     //调用接口
    public static interface  RPCTestInterface extends  VersionedProtocol{
         public static final long versionID = 19L;  
         public String  getUper(String lowerString);
        
    }
    //实现
    public static class  RPCTest implements RPCTestInterface {
        
        @Override
        public String getUper(String lowerString) {
            // TODO Auto-generated method stub
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            return lowerString.toUpperCase();
        }
        @Override
        public long getProtocolVersion(String protocol, long clientVersion)
                throws IOException {
            // TODO Auto-generated method stub
            return versionID;
        }
    }
    @Test
    public void test() {

    }

    
}

     Server端类图：

     

     Server端源代码分析：

  

 

     Client端源代码分析：

    

     RPC类图：

    RPC、Client、Server源代码注释以及分析见附件：https://files.cnblogs.com/serendipity/ipc.rar

源代码片段分析：

1、Server流程

  Server server=RPC.getServer(new RPCTest(), "127.0.0.1", 9813,6, true ,new Configuration());
  
  最终调用的Server构造函数
  public Server(Object instance, Configuration conf, String bindAddress,  int port,
                  int numHandlers, boolean verbose) throws IOException {
      //socket地址、端口、RPC调用过程中传输的函数参数（始终为Invocation）、多少线程来处理请求
      super(bindAddress, port, Invocation.class, numHandlers, conf, classNameBase(instance.getClass().getName()));
      //Server的最终实现的实例，Invocation中包含有方法名、方法参数、参数类型，通过反射instance来调用相应的方法 。
      //Server中有多个handler，每个handler都拥有对instance的引用，其不停扫描callQueue来取得call并处理

      this.instance = instance;
      this.implementation = instance.getClass();
      //是否打印调用日志
      this.verbose = verbose;
    }

   protected Server(String bindAddress, int port, 
                  Class<? extends Writable> paramClass, int handlerCount, 
                  Configuration conf, String serverName) 
    throws IOException {
    this.bindAddress = bindAddress;   
    this.conf = conf;
    this.port = port;
    this.paramClass = paramClass;
    this.handlerCount = handlerCount;
    this.socketSendBufferSize = 0;
    this.maxQueueSize = handlerCount * MAX_QUEUE_SIZE_PER_HANDLER;
    this.callQueue  = new LinkedBlockingQueue<Call>(maxQueueSize); 
    this.maxIdleTime = 2*conf.getInt("ipc.client.connection.maxidletime", 1000);
    this.maxConnectionsToNuke = conf.getInt("ipc.client.kill.max", 10);
    this.thresholdIdleConnections = conf.getInt("ipc.client.idlethreshold", 4000);
    
    // Start the listener here and let it bind to the port
    listener = new Listener();
    this.port = listener.getAddress().getPort();    
    this.rpcMetrics = new RpcMetrics(serverName,
                          Integer.toString(this.port), this);
    this.tcpNoDelay = conf.getBoolean("ipc.server.tcpnodelay", false);


    // Create the responder here
    responder = new Responder();
  }

2、关于Server中的Response，当handler处理完一个call后会把，结果放入到responseQueue中，放回后会判断responseQueue==1 ？，如果是的话，则立刻将结果返回，或者等到Response线程去扫描responseQueue，然后返回client 。

3、socket传输数据格式，每个socket在Server的org.apache.hadoop.ipc.Server.Connection中，其读取socket上数据如下：

//Connection 读取数据
    //每个connection的数据格式：
    //HEADER("hrpc")-CURRENT_VERSION(2)-datalength(ticket的长度)-ticket(前一个length的长度)-datalength(data的长度)-data（长度为前一个长度，data为Invocation的序列化数据）-
    //datalength(data的长度)-data（长度为前一个长度，data为Invocation的序列化数据）-datalength(data的长度)-data（长度为前一个长度，data为Invocation的序列化数据）-......
    //每个连接中HEADER、CURRENT_VERSION、ticket再在连接建立时候发送过来，作为鉴权使用
    //然后即可不断发送data
    public int readAndProcess() throws IOException, InterruptedException {
      System.out.println("-------");
      while (true) {
        /* Read at most one RPC. If the header is not read completely yet
         * then iterate until we read first RPC or until there is no data left.
         */    
        int count = -1;
        //dataLengthBuffer读取每次数据的长度，在连接刚建立时候第一次读取的会是"hrpc"
        if (dataLengthBuffer.remaining() > 0) {
        System.out.println("read data length");
        //其实每次读取的dataLengthBuffer均是"hrpc"的byte[]数组
          count = channelRead(channel, dataLengthBuffer);       
          if (count < 0 || dataLengthBuffer.remaining() > 0) 
            return count;
        }
        if (!versionRead) {
          System.out.println("read versionRead");
          //Every connection is expected to send the header.
          ByteBuffer versionBuffer = ByteBuffer.allocate(1);
          count = channelRead(channel, versionBuffer);
          if (count <= 0) {
            return count;
          }
          int version = versionBuffer.get(0);
          
          dataLengthBuffer.flip();          
          if (!HEADER.equals(dataLengthBuffer) || version != CURRENT_VERSION) {
            //Warning is ok since this is not supposed to happen.
            LOG.warn("Incorrect header or version mismatch from " + 
                     hostAddress + ":" + remotePort +
                     " got version " + version + 
                     " expected version " + CURRENT_VERSION);
            return -1;
          }
          dataLengthBuffer.clear();
          versionRead = true;
          continue;
        }
        
        if (data == null) {
          dataLengthBuffer.flip();
          dataLength = dataLengthBuffer.getInt();
       
          if (dataLength == Client.PING_CALL_ID) {
            dataLengthBuffer.clear();
            return 0;  //ping message
          }
          data = ByteBuffer.allocate(dataLength);
          incRpcCount();  // Increment the rpc count
        }
        
        count = channelRead(channel, data);
        
        if (data.remaining() == 0) {
          dataLengthBuffer.clear();
          data.flip();
          if (headerRead) {
            System.out.println("read data ");
            processData();
            data = null;
            return count;
          } else {
            System.out.println("read data header ticket");
            processHeader();
            headerRead = true;
            data = null;
            continue;
          }
        } 
        return count;
      }
    }

4、handler

   Server在启动后会启动多个线程，每个线程不停的去callQueue中取call，然后调用instance去处理每个请求，处理完将结果放到responseQueue中或者直接通过response发送到客服端 ，每个这样的线程就叫一个handler 。

5、Client的调用：java动态代理 。

   

RPCTestInterface rpcInterface=(RPCTestInterface)RPC.getProxy(RPCTestInterface.class, RPCTestInterface.versionID, nameNodeAddr, new Configuration());
 
/** Construct a client-side proxy object that implements the named protocol,
   * talking to a server at the named address. 
   * 获得相应的代理，获得后会调用getProtocolVersion方法，是以服务器端在你调用实际的方法前会有一个server的方法调用
   * 判断客服端的version和服务端是不是相同？ 不同的话则抛出VersionMismatch异常 。
   * 这一块在调用端控制，其实可以修改代码绕过该判断
   * */
  public static VersionedProtocol getProxy(Class<?> protocol,
      long clientVersion, InetSocketAddress addr, UserGroupInformation ticket,
      Configuration conf, SocketFactory factory) throws IOException {    

    VersionedProtocol proxy =
        (VersionedProtocol) Proxy.newProxyInstance(
            protocol.getClassLoader(), new Class[] { protocol },
            new Invoker(addr, ticket, conf, factory));
    long serverVersion = proxy.getProtocolVersion(protocol.getName(), 
                                                  clientVersion);
    if (serverVersion == clientVersion) {
      return proxy;
    } else {
      throw new VersionMismatch(protocol.getName(), clientVersion, 
                                serverVersion);
    }
  }
//客服端调用的动态代理类
 private static class Invoker implements InvocationHandler {
    private InetSocketAddress address;
    private UserGroupInformation ticket;
    private Client client;
    private boolean isClosed = false;

    public Invoker(InetSocketAddress address, UserGroupInformation ticket, 
                   Configuration conf, SocketFactory factory) {
      this.address = address;
      this.ticket = ticket;
      this.client = CLIENTS.getClient(conf, factory);
    }

    public Object invoke(Object proxy, Method method, Object[] args)
      throws Throwable {
      final boolean logDebug = LOG.isDebugEnabled();
      long startTime = 0;
      if (logDebug) {
        startTime = System.currentTimeMillis();
      }
      ObjectWritable value = (ObjectWritable)
        client.call(new Invocation(method, args), address, ticket);
      if (logDebug) {
        long callTime = System.currentTimeMillis() - startTime;
        LOG.debug("Call: " + method.getName() + " " + callTime);
      }
      return value.get();
    }
    
    /* close the IPC client that's responsible for this invoker's RPCs */ 
    synchronized private void close() {
      if (!isClosed) {
        isClosed = true;
        CLIENTS.stopClient(client);
      }
    }
  }

方法调用时候实际上是对每个接口做动态代理，将其对方法的调用转换到Invoker的invoke(Object proxy, Method method, Object[] args)方法中，在该方法中将方法名字和参数包装成new Invocation(method, args)然后将参数发送 。

6、单个的client调用

   

 public Writable call(Writable param, InetSocketAddress addr, 
                       UserGroupInformation ticket)  
                       throws InterruptedException, IOException {
    Call call = new Call(param);
    Connection connection = getConnection(addr, ticket, call);
    connection.sendParam(call);                 // send the parameter
    synchronized (call) {
      while (!call.done) {
        try {
          call.wait();                           // wait for the result
        } catch (InterruptedException ignored) {}
      }

      if (call.error != null) {
        if (call.error instanceof RemoteException) {
          call.error.fillInStackTrace();
          throw call.error;
        } else { // local exception
          throw wrapException(addr, call.error);
        }
      } else {
        return call.value;
      }
    }
  }

///在Call对象上的唤醒代码：

protected synchronized void callComplete() {
      this.done = true;
      notify();                                 // notify caller
    }

    /** Set the exception when there is an error.
     * Notify the caller the call is done.
     * 
     * @param error exception thrown by the call; either local or remote
     */
    public synchronized void setException(IOException error) {
      this.error = error;
      callComplete();
    }
    
    /** Set the return value when there is no error. 
     * Notify the caller the call is done.
     * 
     * @param value return value of the call.
     */
    public synchronized void setValue(Writable value) {
      this.value = value;
      callComplete();
    }

单个的调用，此时调用线程发送完call后会等待结果返回，是call上的同步 ，其最终由Connection线程收到response数据后将数据set到call内，唤醒该等待线程，代码见上 。

7、多个call的批量调用

     client中除了单个的同步调用，还有批量发送call的方法

  

/** Makes a set of calls in parallel.  Each parameter is sent to the
   * corresponding address.  When all values are available, or have timed out
   * or errored, the collected results are returned in an array.  The array
   * contains nulls for calls that timed out or errored. 
   * 多个请求同时发送，将多个请求打包成ParallelResults，异步发送 
   * 只在ParallelResults上等待
   *  
   *  */
  public Writable[] call(Writable[] params, InetSocketAddress[] addresses)
    throws IOException {
    if (addresses.length == 0) return new Writable[0];

    ParallelResults results = new ParallelResults(params.length);
    synchronized (results) {
      for (int i = 0; i < params.length; i++) {
        ParallelCall call = new ParallelCall(params[i], results, i);
        try {
          Connection connection = getConnection(addresses[i], null, call);
          connection.sendParam(call);             // send each parameter
        } catch (IOException e) {
          // log errors
          LOG.info("Calling "+addresses[i]+" caught: " + 
                   e.getMessage(),e);
          results.size--;                         //  wait for one fewer result
        }
      }
      while (results.count != results.size) {
        try {
          results.wait();                    // wait for all results
        } catch (InterruptedException e) {}
      }

      return results.values;
    }
  }

其将所有的call打包成ParallelResults，然后在ParallelResults上等待，但单个call是异步socket发送的，不需要等待签个call的返回 。

8、