Hadoop源码分析之一(RPC机制之Server)

比眉伴天荒 2022-03-20 02:05 176阅读 0赞

Rpc源码分析比较给力，详细：

### [源码级强力分析hadoop的RPC机制][hadoop_RPC]：http://weixiaolu.iteye.com/blog/1504898 ###

http://www.linuxidc.com/Linux/2013-01/77251p2.htm

网上已经有很多关于[Hadoop][]源码分析的好文，在这记录的目的是把自己在看Hadoop源码时自己的一些体会，方便日后查询巩固。

相关阅读：Hadoop源码分析之二(RPC机制之Call处理) [http://www.linuxidc.com/Linux/2013-01/77252.htm][http_www.linuxidc.com_Linux_2013-01_77252.htm]

想了解Hadoop是如何工作的，首先需要了解Hadoop RPC原理。Hadoop提供了一个统一的RPC机制来处理client-namenode, namenode-dataname,client-dataname之间的通信。整个机制中主要运用了如下技术：

1.  Java dynamic proxy：主要用于处理client的代理来统一向Server端发送信息；
2.  [Google Protocal Buffer][]：主要用于把Request和Response序列化和反序列化成byte进行通信；
3.  Java NIO：主要用于RPC Client与Server之间的通信；

下面主要介绍RPC Server主要的实现逻辑：

![ctrl+鼠标左键原大小][ctrl]

如上图Server端主要Listener, Responer Thread和Handler, Reader Thread Pool。

1.  Listener Thread：Server端会启一个Listener线程主要用于监听Client发送过来的Request，Listener主要完成创建一个Connection Object并启动一组Reader Thread Pool，并把Connection通过NIO的SelectionKey传递给Reader，这样就解决了Listener单线程带来的可能的性能瓶颈，因为Listener只作了一层转发；
2.  Reader Thread Pool：主要用于读取Listener传过来的Connection，并调用Connection的readAndProcess方法来读取Request，并封装成一个Call放到Call Queue中；
3.  Hanlder Thread Pool：Server会启动一组线程组来处理Call Queue中Call，并把处理的Respone中放到response queue中，Hanlder怎么处理Call会在后续介绍；
4.  Responder Thread：主要处理response queue中的response，并把response发送给client，如果当前response queue为空，在加了新的response时会马上发送给client端，不会通过responer thread来发送。
5.  下面主要结合实际代码来介绍一下[Hadoop][] RPC Server端整个的处理过程：（整个核心逻辑都在org.apache.hadoop.ipc.Server这个类中实现）
    
    首先在Server的start方法中启用相关的线程，start方法会在相关的Server中调用，如NameNode, DataNode等
    
      public synchronized void start() \{  
        responder.start();  
        listener.start();  
        handlers = new Handler\[handlerCount\];  
          
        for (int i = 0; i < handlerCount; i++) \{  
          handlers\[i\] = new Handler(i);  
          handlers\[i\].start();  
        \}  
      \}
    
    如上可以看到启动了listener,responer和handler threand pool，可以通过ipc.server.handler.queue.size来配置handler的个数，默认为100个。  
    在Server中有一个Listener的内部类，Listener的创造函数中会启动相应的Reader Thread Pool，可以通过ipc.server.read.threadpool.size来配置Reader的线程数，默认为1个。Listener会监听OP\_ACCEPT事件，并在doAccpet中创建一个Connection传递给Reader，如下：
    
    while ((channel = server.accept()) != null) \{  
            channel.configureBlocking(false);  
            channel.socket().setTcpNoDelay(tcpNoDelay);  
            //get one reader instance for reader pool  
            Reader reader = getReader();  
            try \{  
          //lock this reader  
              reader.startAdd();  
              SelectionKey readKey = reader.registerChannel(channel);  
              //build Connection instance  
              c = new Connection(readKey, channel, Time.now());  
              readKey.attach(c);  
              synchronized (connectionList) \{  
                connectionList.add(numConnections, c);  
                numConnections++;  
              \}      
            \} finally \{  
              //unlock this reader  
              reader.finishAdd();   
            \}  
          \}  
        \}
    
    在Reader中主要通用Connection来读取Requeset并封装成Call，读取Request分为2种处理，一种为权限读取，另一种为普通读取，具体的实现可以查看hadoop源码，这里不粘出。但是这里会介绍一下整个request在整个传送过程中的结构，request由head和conent组成。
    
    Head的组成如下：
    
    \* +----------------------------------+  
    \* | "hrpc" 4 bytes |   
    \* +----------------------------------+  
    \* | Version (1 bytes) |   
    \* +----------------------------------+  
    \* | Authmethod (1 byte) |   
    \* +----------------------------------+  
    \* | IpcSerializationType (1 byte) |   
    \* +----------------------------------+
    
    conent主要有4 bytes的是date length和具体的data byte array组成。
    
    Handler怎么处理具体的Call会在后续具体介绍，这里不做具体展开。
    
    Responder主要处理Call产生的Response并通过NIO传输给具体的client， 使用的socket channel与request为同一个，Responder会做相应的check，如下：
    
     
    
        //check select key channel is same as call.connection.channel  
        if (key.channel() != call.connection.channel) \{  
            throw new IOException("doAsyncWrite: bad channel");  
        \}
    
     
    
        // Extract the first call  
        call = responseQueue.removeFirst();  
        SocketChannel channel = call.connection.channel;  
        // Using call channel to send as much data as we can in the non-blocking fashion  
        int numBytes = channelWrite(channel, call.rpcResponse);
    
     
    
    整个Hadoop RPC Server处理的机制就介绍到这里，后续会介绍call怎么处理到具体的处理及Client如何向Server发送信息。

[hadoop_RPC]: http://weixiaolu.iteye.com/blog/1504898
[Hadoop]: http://www.linuxidc.com/topicnews.aspx?tid=13
[http_www.linuxidc.com_Linux_2013-01_77252.htm]: http://www.linuxidc.com/Linux/2013-01/77252.htm
[Google Protocal Buffer]: http://code.google.com/p/protobuf/
[ctrl]: /images/20220320/21e6c1edcaf943898652685a798bbc31.png