Hadoop Rpc簡單實作

一、Hadoop RPC總體架構

• 序列化層
  序列化作用主要還是將結構化資料物件轉換成位元組流用于網路傳輸或寫入持久存盤，
  在RPC中，主要是將用戶請求的引數或者服務器應答轉化成位元組流跨機器傳輸，
• 函式呼叫層
  作用：定位所需呼叫的函式并執行函式，
  依賴：Java的反射與Java的動態代理模式來實作
• 網路傳輸層
  作用：用于描述Client與Server之間的訊息格式，
  依賴：Hadoop RPC 依賴基于TCP/IP協議中的Socket機制
• 服務器端處理框架
  作用：抽象為網路IO模型，用于述client與server間資訊互動方式，
  網路IO模型：阻塞式IO、NIO、事件驅動IO等
  Hadoop RPC：基于Reactor設計模式的事件驅動IO模型，

二、Hadoop Rpc特點

• 透明性
  這是所有RPC框架最根本的特點，即當用戶在一臺計算機的程式呼叫另外一臺計算機上的子程式時，用戶自身不應感覺到其間涉及跨機器間的通信，而是感覺像是在執行一個本地呼叫，
• 高性能
  Hadoop各個系統（如HDFS、YARN、MapReduce等）均采用了Master/Slave結構，其中，Master實際上是一個RPC Server，它負責處理集群中所有Slave發送的服務請求，為了保證Master的并發處理能力，RPCServer應是一個高性能服務器，能夠高效地處理來自多個Client的并發RPC請求，
• 可控性
  相對于RMI(remote method invocation)而言，Hadoop RPC具有輕量級、可控性等優點，即用戶程式呼叫RPC介面可控的地方比較RMI多些，

三、實作步驟

• 創建Maven專案，添加maven依賴（maven有問題，可以直接匯入hadoop相關依賴包）

  	<dependencies>
  	        <dependency>
  	            <groupId>org.apache.hadoop</groupId>
  	            <artifactId>hadoop-common</artifactId>
  	            <version>2.7.3</version>
  	        </dependency>
  	        <dependency>
  	            <groupId>org.apache.hadoop</groupId>
  	            <artifactId>hadoop-hdfs</artifactId>
  	            <version>2.7.3</version>
  	        </dependency>
  	        <dependency>
  	            <groupId>org.apache.hadoop</groupId>
  	            <artifactId>hadoop-mapreduce-client-common</artifactId>
  	            <version>2.7.3</version>
  	        </dependency>
  	        <dependency>
  	            <groupId>org.apache.hadoop</groupId>
  	            <artifactId>hadoop-mapreduce-client-core</artifactId>
  	            <version>2.7.3</version>
  	        </dependency>
  

• 定義介面協議

  /**
   * 1、定義RPC介面協議，添加自定義方法
   * 必須繼承Hadoop提供的介面VersionedProtocol
   */
  public interface RPCProtocol {
      //RPC client 和 server 之間必須使用相同的版本的協議才能進行正常通信
      public static final long versionID = 1L;
      //自定義方法
      public String echo(String value) throws Exception;
      public int add(int v1, int v2) throws Exception;
  }
  

• 定義介面協議實作類

  import org.apache.hadoop.ipc.ProtocolSignature;
  import java.io.IOException;
  
  /**
   * 2、定義一個Java類，實作自定義的RCP介面
   */
  public class RPCProtocolImplement implements RPCProtocol{
      public long getProtocolVersion(String protocol, long clientVersion) throws IOException {
          System.out.println("===getProtocolVersion被呼叫===protocol=" + protocol + "\t clientVersion=" + clientVersion);
          return RPCProtocol.versionID;
      }
      public ProtocolSignature getProtocolSignature(String protocol, long clientVersion, int clientMethodsHash) throws IOException {
          System.out.println("===getProtocolSignature被呼叫===protocol=" + protocol + ",clientVersion=" + clientVersion + ",clientMethodsHash=" + clientMethodsHash);
          return new ProtocolSignature(RPCProtocol.versionID,null);
      }
  
      public String echo(String value) throws Exception {
          System.out.println("好的，我已收到你的資訊");
          return value;
      }
  
      public int add(int v1, int v2) throws Exception {
          System.out.println("正在幫您計算，稍等片刻……");
          int sum = v1 + v2;
          System.out.println("計算完畢，結果為：" + sum);
          return sum;
      }
  }
  

• 定義服務端（Server）

  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.ipc.RPC;
  
  import java.io.IOException;
  
  /**
   * 3、構造RPC服務啟動類
   */
  public class RPCServer {
      public static void main(String[] args) throws IOException {
          RPC.Server server = new RPC.Builder(new Configuration())
                  .setProtocol(RPCProtocol.class)
                  .setInstance(new RPCProtocolImplement())
                  .setBindAddress("localhost")
                  .setPort(8001).setNumHandlers(1).build();
          server.start();
      }
  }
  

• 定義客戶端Client

  import org.apache.hadoop.conf.Configuration;
  import org.apache.hadoop.ipc.RPC;
  
  import java.net.InetSocketAddress;
  
  /**
   * 4、定義RPC客戶端
   */
  public class RPCClient {
      public static void main(String[] args) throws Exception {
          RPCProtocol client = RPC.getProxy(RPCProtocol.class,
                  RPCProtocol.versionID,
                  new InetSocketAddress("localhost",8001),
                  new Configuration());
          String echo = client.echo("rpc服務，請幫我計算一下：");
          System.out.println(echo);
          int sum = client.add(400, 300);
          System.out.println("收到rpc服務的計算結果:" + sum);
          // 停止客戶端
          RPC.stopProxy(client);
      }
  }
  

四、運行及實驗結果

• 啟動RPCServer服務端服務（先啟動）
  
• 啟動RPCClient客戶端呼叫服務
• 運行結果如下

  • 客戶端Client控制臺會看到如下日志：
    

  • 服務端Server控制臺會看到如下日志：