偷偷摘套内射激情视频,久久精品99国产国产精,中文字幕无线乱码人妻,中文在线中文a,性爽19p

Hadoop RPC通信Client客戶端的流程分析

作者：Android路上的人
安全 云安全 數(shù)據(jù)庫 Hadoop
Hadoop的RPC的通信與其他系統(tǒng)的RPC通信不太一樣，作者針對Hadoop的使用特點，專門的設計了一套RPC框架，這套框架個人感覺還是有點小復雜的。

[[124009]]

Hadoop的RPC的通信與其他系統(tǒng)的RPC通信不太一樣，作者針對Hadoop的使用特點，專門的設計了一套RPC框架，這套框架個人感覺還是 有點小復雜的。所以我打算分成Client客戶端和Server服務端2個模塊做分析。如果你對RPC的整套流程已經(jīng)非常了解的前提下，對于Hadoop 的RPC，你也一定可以非常迅速的了解的。OK，下面切入正題。

Hadoop的RPC的相關代碼都在org.apache.hadoop.ipc的包下，首先RPC的通信必須遵守許多的協(xié)議，其中最最基本的協(xié)議即使如下：

  1. /** 
  2.  * Superclass of all protocols that use Hadoop RPC. 
  3.  * Subclasses of this interface are also supposed to have 
  4.  * a static final long versionID field. 
  5.  * Hadoop RPC所有協(xié)議的基類，返回協(xié)議版本號 
  6.  */ 
  7. public interface VersionedProtocol { 
  8.    
  9.   /** 
  10.    * Return protocol version corresponding to protocol interface. 
  11.    * @param protocol The classname of the protocol interface 
  12.    * @param clientVersion The version of the protocol that the client speaks 
  13.    * @return the version that the server will speak 
  14.    */ 
  15.   public long getProtocolVersion(String protocol,  
  16.                                  long clientVersion) throws IOException; 
  17. } 

他是所有協(xié)議的基類，他的下面還有一堆的子類，分別對應于不同情況之間的通信，下面是一張父子類圖：

顧名思義，只有客戶端和服務端遵循相同的版本號，才能進行通信。

RPC客戶端的所有相關操作都被封裝在了一個叫Client.java的文件中:

  1. /** A client for an IPC service.  IPC calls take a single {@link Writable} as a 
  2.  * parameter, and return a {@link Writable} as their value.  A service runs on 
  3.  * a port and is defined by a parameter class and a value class. 
  4.  * RPC客戶端類 
  5.  * @see Server 
  6.  */ 
  7. public class Client { 
  8.    
  9.   public static final Log LOG = 
  10.     LogFactory.getLog(Client.class); 
  11.   //客戶端到服務端的連接 
  12.   private Hashtable<ConnectionId, Connection> connections = 
  13.     new Hashtable<ConnectionId, Connection>(); 
  14.  
  15.   //回調(diào)值類 
  16.   private Class<? extends Writable> valueClass;   // class of call values 
  17.   //call回調(diào)id的計數(shù)器 
  18.   private int counter;                            // counter for call ids 
  19.   //原子變量判斷客戶端是否還在運行 
  20.   private AtomicBoolean running = new AtomicBoolean(true); // if client runs 
  21.   final private Configuration conf; 
  22.  
  23.   //socket工廠，用來創(chuàng)建socket 
  24.   private SocketFactory socketFactory;           // how to create sockets 
  25.   private int refCount = 1; 
  26.   ...... 

從代碼中明顯的看到，這里存在著一個類似于connections連接池的東西，其實這暗示著連接是可以被復用的，在hashtable中，與每個Connecttion連接的對應的是一個ConnectionId，顯然這里不是一個Long類似的數(shù)值:

  1. /** 
  2.     * This class holds the address and the user ticket. The client connections 
  3.     * to servers are uniquely identified by <remoteAddress, protocol, ticket> 
  4.     * 連接的唯一標識，主要通過<遠程地址，協(xié)議類型，用戶組信息> 
  5.     */ 
  6.    static class ConnectionId { 
  7.      //遠程的socket地址 
  8.      InetSocketAddress address; 
  9.      //用戶組信息 
  10.      UserGroupInformation ticket; 
  11.      //協(xié)議類型 
  12.      Class<?> protocol; 
  13.      private static final int PRIME = 16777619; 
  14.      private int rpcTimeout; 
  15.      private String serverPrincipal; 
  16.      private int maxIdleTime; //connections will be culled if it was idle for  
  17.      //maxIdleTime msecs 
  18.      private int maxRetries; //the max. no. of retries for socket connections 
  19.      private boolean tcpNoDelay; // if T then disable Nagle's Algorithm 
  20.      private int pingInterval; // how often sends ping to the server in msecs 
  21.      .... 

這里用了3個屬性組成唯一的標識屬性，為了保證可以進行ID的復用，所以作者對ConnectionId的equal比較方法和hashCode 進行了重寫:

  1. /** 
  2.       * 作者重寫了equal比較方法，只要成員變量都想等也就想到了 
  3.       */ 
  4.      @Override 
  5.      public boolean equals(Object obj) { 
  6.        if (obj == this) { 
  7.          return true; 
  8.        } 
  9.        if (obj instanceof ConnectionId) { 
  10.          ConnectionId that = (ConnectionId) obj; 
  11.          return isEqual(this.address, that.address) 
  12.              && this.maxIdleTime == that.maxIdleTime 
  13.              && this.maxRetries == that.maxRetries 
  14.              && this.pingInterval == that.pingInterval 
  15.              && isEqual(this.protocol, that.protocol) 
  16.              && this.rpcTimeout == that.rpcTimeout 
  17.              && isEqual(this.serverPrincipal, that.serverPrincipal) 
  18.              && this.tcpNoDelay == that.tcpNoDelay 
  19.              && isEqual(this.ticket, that.ticket); 
  20.        } 
  21.        return false; 
  22.      } 
  23.       
  24.      /** 
  25.       * 重寫了hashCode的生成規(guī)則，保證不同的對象產(chǎn)生不同的hashCode值 
  26.       */ 
  27.      @Override 
  28.      public int hashCode() { 
  29.        int result = 1; 
  30.        result = PRIME * result + ((address == null) ? 0 : address.hashCode()); 
  31.        result = PRIME * result + maxIdleTime; 
  32.        result = PRIME * result + maxRetries; 
  33.        result = PRIME * result + pingInterval; 
  34.        result = PRIME * result + ((protocol == null) ? 0 : protocol.hashCode()); 
  35.        result = PRIME * rpcTimeout; 
  36.        result = PRIME * result 
  37.            + ((serverPrincipal == null) ? 0 : serverPrincipal.hashCode()); 
  38.        result = PRIME * result + (tcpNoDelay ? 1231 : 1237); 
  39.        result = PRIME * result + ((ticket == null) ? 0 : ticket.hashCode()); 
  40.        return result; 
  41.      } 

這樣就能保證對應同類型的連接就能夠完全復用了，而不是僅僅憑借引用的關系判斷對象是否相等，這里就是一個不錯的設計了。

與連接Id對應的就是Connection了，它里面維護是一下的一些變量；

  1.  /** Thread that reads responses and notifies callers.  Each connection owns a 
  2.   * socket connected to a remote address.  Calls are multiplexed through this 
  3.   * socket: responses may be delivered out of order. */ 
  4.  private class Connection extends Thread { 
  5. //所連接的服務器地址 
  6.    private InetSocketAddress server;             // server ip:port 
  7.    //服務端的krb5的名字，與安全方面相關 
  8.    private String serverPrincipal;  // server's krb5 principal name 
  9.    //連接頭部，內(nèi)部包含了，所用的協(xié)議，客戶端用戶組信息以及驗證的而方法 
  10.    private ConnectionHeader header;              // connection header 
  11.    //遠程連接ID  
  12.    private final ConnectionId remoteId;                // connection id 
  13.    //連接驗證方法 
  14.    private AuthMethod authMethod; // authentication method 
  15.    //下面3個變量都是安全方面的 
  16.    private boolean useSasl; 
  17.    private Token<? extends TokenIdentifier> token; 
  18.    private SaslRpcClient saslRpcClient; 
  19.     
  20.    //下面是一組socket通信方面的變量 
  21.    private Socket socket = null;                 // connected socket 
  22.    private DataInputStream in; 
  23.    private DataOutputStream out; 
  24.    private int rpcTimeout; 
  25.    private int maxIdleTime; //connections will be culled if it was idle for 
  26.         //maxIdleTime msecs 
  27.    private int maxRetries; //the max. no. of retries for socket connections 
  28.    //tcpNoDelay可設置是否阻塞模式 
  29.    private boolean tcpNoDelay; // if T then disable Nagle's Algorithm 
  30.    private int pingInterval; // how often sends ping to the server in msecs 
  31.     
  32.    // currently active calls 當前活躍的回調(diào)，一個連接 可能會有很多個call回調(diào) 
  33.    private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>(); 
  34.    //最后一次IO活動通信的時間 
  35.    private AtomicLong lastActivity = new AtomicLong();// last I/O activity time 
  36.    //連接關閉標記 
  37.    private AtomicBoolean shouldCloseConnection = new AtomicBoolean();  // indicate if the connection is closed 
  38.    private IOException closeException; // close reason 
  39.    ..... 

里面維護了大量的和連接通信相關的變量，在這里有一個很有意思的東西connectionHeader，連接頭部，里面的數(shù)據(jù)時為了在通信最開始的時候被使用:

  1. class ConnectionHeader implements Writable { 
  2.   public static final Log LOG = LogFactory.getLog(ConnectionHeader.class); 
  3.    
  4.   //客戶端和服務端通信的協(xié)議名稱 
  5.   private String protocol; 
  6.   //客戶端的用戶組信息 
  7.   private UserGroupInformation ugi = null; 
  8.   //驗證的方式，關系到寫入數(shù)據(jù)的時的格式 
  9.   private AuthMethod authMethod; 
  10.   ..... 

起到標識驗證的作用。一個Client類的基本結構我們基本可以描繪出來了，下面是完整的類關系圖:

在上面這幅圖中，你肯定會發(fā)現(xiàn)我少了一個很關鍵的類了，就是Call回調(diào)類。Call回調(diào)在很多異步通信中是經(jīng)常出現(xiàn)的。因為在通信過程中，當一個對象通 過網(wǎng)絡發(fā)送請求給另外一個對象的時候，如果采用同步的方式，會一直阻塞在那里，會帶來非常不好的效率和體驗的，所以很多時候，我們采用的是一種叫回調(diào)接口 的方式。在這期間，用戶可以繼續(xù)做自己的事情。所以同樣的Call這個概念當然也是適用在Hadoop RPC中。在Hadoop的RPC的核心調(diào) 用原理， 簡單的說，就是我把parame參數(shù)序列化到一個對象中，通過參數(shù)的形式把對象傳入，進行RPC通信，最后服務端把處理好的結果值放入call對象，在返 回給客戶端，也就是說客戶端和服務端都是通過Call對象進行操作，Call里面存著，請求的參數(shù)，和處理后的結構值2個變量。通過Call對象的封裝， 客戶單實現(xiàn)了完美的無須知道細節(jié)的調(diào)用。下面是Call類的類按時：

  1. /** A call waiting for a value. */ 
  2. //客戶端的一個回調(diào) 
  3. private class Call { 
  4. /回調(diào)ID 
  5.   int id;                                       // call id 
  6.   //被序列化的參數(shù) 
  7.   Writable param;                               // parameter 
  8.   //返回值 
  9.   Writable value;                               // value, null if error 
  10.   //出錯時返回的異常 
  11.   IOException error;                            // exception, null if value 
  12.   //回調(diào)是否已經(jīng)被完成 
  13.   boolean done;                                 // true when call is done 
  14.   .... 

看到這個Call回調(diào)類，也許你慢慢的會明白Hadoop RPC的一個基本原型了，這些Call當然是存在于某個連接中的，一個連接可能會發(fā)生多個回調(diào)，所以在Connection中維護了calls列表：

  1. private class Connection extends Thread { 
  2.   .... 
  3.   // currently active calls 當前活躍的回調(diào)，一個連接 可能會有很多個call回調(diào) 
  4.   private Hashtable<Integer, Call> calls = new Hashtable<Integer, Call>(); 

作者在設計Call類的時候，比較聰明的考慮一種并發(fā)情況下的Call調(diào)用，所以為此設計了下面這個Call的子類，就是專門用于短時間內(nèi)的瞬間Call調(diào)用:

  1. /** Call implementation used for parallel calls. */ 
  2. /** 繼承自Call回調(diào)類，可以并行的使用，通過加了index下標做Call的區(qū)分 */ 
  3. private class ParallelCall extends Call { 
  4. /每個ParallelCall并行的回調(diào)就會有對應的結果類 
  5.   private ParallelResults results; 
  6.   //index作為Call的區(qū)分 
  7.   private int index; 
  8.   .... 

如果要查找值，就通過里面的ParallelCall查找，原理是根據(jù)index索引:

  1.  /** Result collector for parallel calls. */ 
  2.  private static class ParallelResults { 
  3. //并行結果類中擁有一組返回值，需要ParallelCall的index索引匹配 
  4.    private Writable[] values; 
  5.    //結果值的數(shù)量 
  6.    private int size; 
  7.    //values中已知的值的個數(shù) 
  8.    private int count; 
  9.  
  10.    ..... 
  11.  
  12.    /** Collect a result. */ 
  13.    public synchronized void callComplete(ParallelCall call) { 
  14.      //將call中的值賦給result中 
  15.      values[call.index] = call.value;            // store the value 
  16.      count++;                                    // count it 
  17.      //如果計數(shù)的值等到最終大小，通知caller 
  18.      if (count == size)                          // if all values are in 
  19.        notify();                                 // then notify waiting caller 
  20.    } 
  21.  } 

因為Call結構集是這些并發(fā)Call共有的，所以用的是static變量，都存在在了values數(shù)組中了，只有所有的并發(fā)Call都把值取出來了，才 算回調(diào)成功，這個是個非常細小的輔助設計，這個在有些書籍上并沒有多少提及。下面我們看看一般Call回調(diào)的流程，正如剛剛說的，最終客戶端看到的形式就 是，傳入?yún)?shù)，獲得結果，忽略內(nèi)部一切邏輯，這是怎么做到的呢,答案在下面:

在執(zhí)行之前，你會先得到ConnectionId：

  1. public Writable call(Writable param, InetSocketAddress addr,  
  2.                        Class<?> protocol, UserGroupInformation ticket, 
  3.                        int rpcTimeout) 
  4.                        throws InterruptedException, IOException { 
  5.     ConnectionId remoteId = ConnectionId.getConnectionId(addr, protocol, 
  6.         ticket, rpcTimeout, conf); 
  7.     return call(param, remoteId); 
  8.   } 

接著才是主流程：

  1. public Writable call(Writable param, ConnectionId remoteId)   
  2.                        throws InterruptedException, IOException { 
  3.     //根據(jù)參數(shù)構造一個Call回調(diào) 
  4.     Call call = new Call(param); 
  5.     //根據(jù)遠程ID獲取連接 
  6.     Connection connection = getConnection(remoteId, call); 
  7.     //發(fā)送參數(shù) 
  8.     connection.sendParam(call);                 // send the parameter 
  9.     boolean interrupted = false; 
  10.     synchronized (call) { 
  11.       //如果call.done為false，就是Call還沒完成 
  12.       while (!call.done) { 
  13.         try { 
  14.           //等待遠端程序的執(zhí)行完畢 
  15.           call.wait();                           // wait for the result 
  16.         } catch (InterruptedException ie) { 
  17.           // save the fact that we were interrupted 
  18.           interrupted = true; 
  19.         } 
  20.       } 
  21.  
  22.       //如果是異常中斷，則終止當前線程 
  23.       if (interrupted) { 
  24.         // set the interrupt flag now that we are done waiting 
  25.         Thread.currentThread().interrupt(); 
  26.       } 
  27.  
  28.       //如果call回到出錯，則返回call出錯信息 
  29.       if (call.error != null) { 
  30.         if (call.error instanceof RemoteException) { 
  31.           call.error.fillInStackTrace(); 
  32.           throw call.error; 
  33.         } else { // local exception 
  34.           // use the connection because it will reflect an ip change, unlike 
  35.           // the remoteId 
  36.           throw wrapException(connection.getRemoteAddress(), call.error); 
  37.         } 
  38.       } else { 
  39.         //如果是正常情況下，返回回調(diào)處理后的值 
  40.         return call.value; 
  41.       } 
  42.     } 
  43.   } 

在這上面的操作步驟中，重點關注2個函數(shù)，獲取連接操作，看看人家是如何保證連接的復用性的:

  1. private Connection getConnection(ConnectionId remoteId, 
  2.                                    Call call) 
  3.                                    throws IOException, InterruptedException { 
  4.     ..... 
  5.     /* we could avoid this allocation for each RPC by having a   
  6.      * connectionsId object and with set() method. We need to manage the 
  7.      * refs for keys in HashMap properly. For now its ok. 
  8.      */ 
  9.     do { 
  10.       synchronized (connections) { 
  11.         //從connection連接池中獲取連接，可以保證相同的連接ID可以復用 
  12.         connection = connections.get(remoteId); 
  13.         if (connection == null) { 
  14.           connection = new Connection(remoteId); 
  15.           connections.put(remoteId, connection); 
  16.         } 
  17.       } 
  18.     } while (!connection.addCall(call)); 

有點單例模式的味道哦，還有一個方法叫sendParam發(fā)送參數(shù)方法:

  1. public void sendParam(Call call) { 
  2.   if (shouldCloseConnection.get()) { 
  3.     return; 
  4.   } 
  5.  
  6.   DataOutputBuffer d=null; 
  7.   try { 
  8.     synchronized (this.out) { 
  9.       if (LOG.isDebugEnabled()) 
  10.         LOG.debug(getName() + " sending #" + call.id); 
  11.        
  12.       //for serializing the 
  13.       //data to be written 
  14.       //將call回調(diào)中的參數(shù)寫入到輸出流中，傳向服務端 
  15.       d = new DataOutputBuffer(); 
  16.       d.writeInt(call.id); 
  17.       call.param.write(d); 
  18.       byte[] data = d.getData(); 
  19.       int dataLength = d.getLength(); 
  20.       out.writeInt(dataLength);      //first put the data length 
  21.       out.write(data, 0, dataLength);//write the data 
  22.       out.flush(); 
  23.     } 
  24.     .... 

代碼只發(fā)送了Call的id，和請求參數(shù)，并沒有把所有的Call的內(nèi)容都扔出去了，一定是為了減少數(shù)據(jù)量的傳輸，這里還把數(shù)據(jù)的長度寫入了，這是為了方 便服務端準確的讀取到不定長的數(shù)據(jù)。這服務端中間的處理操作不是今天討論的重點。Call的執(zhí)行過程就是這樣。那么Call是如何被調(diào)用的呢，這又要重新 回到了Client客戶端上去了，Client有一個run()函數(shù)，所有的操作都是始于此的；

  1. public void run() { 
  2.   if (LOG.isDebugEnabled()) 
  3.     LOG.debug(getName() + ": starting, having connections "  
  4.         + connections.size()); 
  5.  
  6.   //等待工作，等待請求調(diào)用 
  7.   while (waitForWork()) {//wait here for work - read or close connection 
  8.     //調(diào)用完請求，則立即獲取回復 
  9.     receiveResponse(); 
  10.   } 
  11.    
  12.   close(); 
  13.    
  14.   if (LOG.isDebugEnabled()) 
  15.     LOG.debug(getName() + ": stopped, remaining connections " 
  16.         + connections.size()); 
  17. } 

操作很簡單，程序一直跑著，有請求，處理請求，獲取請求，沒有請求，就死等。

  1. private synchronized boolean waitForWork() { 
  2.       if (calls.isEmpty() && !shouldCloseConnection.get()  && running.get())  { 
  3.         long timeout = maxIdleTime- 
  4.               (System.currentTimeMillis()-lastActivity.get()); 
  5.         if (timeout>0) { 
  6.           try { 
  7.             wait(timeout); 
  8.           } catch (InterruptedException e) {} 
  9.         } 
  10.       } 
  11.       .... 

獲取回復的操作如下:

  1. /* Receive a response. 
  2.      * Because only one receiver, so no synchronization on in. 
  3.      * 獲取回復值 
  4.      */ 
  5.     private void receiveResponse() { 
  6.       if (shouldCloseConnection.get()) { 
  7.         return; 
  8.       } 
  9.       //更新最近一次的call活動時間 
  10.       touch(); 
  11.        
  12.       try { 
  13.         int id = in.readInt();                    // try to read an id 
  14.  
  15.         if (LOG.isDebugEnabled()) 
  16.           LOG.debug(getName() + " got value #" + id); 
  17.  
  18.         //從獲取call中取得相應的call 
  19.         Call call = calls.get(id); 
  20.  
  21.         //判斷該結果狀態(tài) 
  22.         int state = in.readInt();     // read call status 
  23.         if (state == Status.SUCCESS.state) { 
  24.           Writable value = ReflectionUtils.newInstance(valueClass, conf); 
  25.           value.readFields(in);                 // read value 
  26.           call.setValue(value); 
  27.           calls.remove(id); 
  28.         } else if (state == Status.ERROR.state) { 
  29.           call.setException(new RemoteException(WritableUtils.readString(in), 
  30.                                                 WritableUtils.readString(in))); 
  31.           calls.remove(id); 
  32.         } else if (state == Status.FATAL.state) { 
  33.           // Close the connection 
  34.           markClosed(new RemoteException(WritableUtils.readString(in),  
  35.                                          WritableUtils.readString(in))); 
  36.         } 
  37.         ..... 
  38.       } catch (IOException e) { 
  39.         markClosed(e); 
  40.       } 
  41.     } 

從之前維護的Call列表中取出，做判斷。Client本身的執(zhí)行流程比較的簡單:

Hadoop RPC客戶端的通信模塊的部分大致就是我上面的這個流程，中間其實還忽略了很多的細節(jié)，大家學習的時候，針對源碼會有助于更好的理解，Hadoop RPC的服務端的實現(xiàn)更加復雜，所以建議采用分模塊的學習或許會更好一點。

本文出自：http://blog.csdn.net/Androidlushangderen/article/details/41751133

責任編輯：Ophira 來源： Android路上的人的博客
hadoop框架線程網(wǎng)絡遠程連接
相關推薦

2009-03-13 14:44:55

客戶端C#Oracle

2009-08-01 22:47:58

2009-08-21 15:59:22

服務端與客戶端通信

2009-08-21 16:14:52

服務端與客戶端通信

2021-09-22 15:46:29

虛擬桌面瘦客戶端胖客戶端

2021-06-22 15:06:13

Redis客戶端 Redis-clie

2021-07-16 06:56:50

Nacos注冊源碼

2010-03-18 17:47:07

Java 多客戶端通信

2010-03-19 12:14:13

無線AP Client

2010-05-23 23:36:38

思科統(tǒng)一通信

2011-03-21 14:53:36

Nagios監(jiān)控Linux

2011-04-06 14:24:20

Nagios監(jiān)控Linux

2011-08-17 10:10:59

2009-11-09 15:49:01

WCF異步調(diào)用

2011-10-31 13:42:46

Android客戶端人人網(wǎng)

2021-10-19 08:58:48

Java 語言 Java 基礎

2013-04-03 09:27:42

2010-05-31 10:11:32

瘦客戶端

2011-10-26 13:17:05

2011-03-24 13:00:31

配置nagios客戶端
點贊
收藏

51CTO技術棧公眾號