在Spark中很多地方都涉及網路通訊，比如Spark各個元件間的訊息互通、使用者檔案與Jar包的上傳、節點間的Shuffle過程、Block資料的複製與備份等。在Spark 0.x.x與Spark 1.x.x版本中，元件間的訊息通訊主要藉助於Akka[1]，使用Akka可以輕鬆的構建強有力的高併發與分散式應用。但是Akka在Spark 2.0.0版本中被移除了，Spark官網文件對此的描述為：“Akka的依賴被移除了，因此使用者可以使用任何版本的Akka來程式設計了。”Spark團隊的決策者或許認為對於Akka具體版本的依賴，限制了使用者對於Akka不同版本的使用。儘管如此，筆者依然認為Akka是一款非常優秀的開源分散式系統，我參與的一些Java Application或者Java Web就利用Akka的豐富特性實現了分散式一致性、最終一致性以及分散式事務等分散式環境面對的問題。在Spark 1.x.x版本中，使用者檔案與Jar包的上傳採用了由Jetty[2]

       我們先來看看RPC框架的基本架構，如圖1所示。

圖1       Spark內建RPC框架的基本架構

TransportContext內部包含傳輸上下文的配置資訊TransportConf和對客戶端請求訊息進行處理的RpcHandler。TransportConf在建立TransportClientFactory和TransportServer時都是必須的，而RpcHandler只用於建立TransportServer。TransportClientFactory是RPC客戶端的工廠類。TransportServer是RPC服務端的實現。圖中記號的含義如下：

記號①表示通過呼叫TransportContext的createClientFactory方法建立傳輸客戶端工廠TransportClientFactory的例項。在構造TransportClientFactory的例項時，還會傳遞客戶端載入程式TransportClientBootstrap的列表。此外，TransportClientFactory內部還存在針對每個Socket地址的連線池ClientPool，這個連線池快取的定義如下：

  private final ConcurrentHashMap<SocketAddress, ClientPool> connectionPool;
 

ClientPool的型別定義如下：

  private static class ClientPool {
    TransportClient[] clients;
    Object[] locks;

    ClientPool(int size) {
      clients = new TransportClient[size];
      locks = new Object[size];
      for (int i = 0; i < size; i++) {
        locks[i] = new Object();
      }
    }
  }

由此可見，ClientPool實際是由TransportClient的陣列構成，而locks陣列中的Object與clients陣列中的TransportClient按照陣列索引一一對應，通過對每個TransportClient分別採用不同的鎖，降低併發情況下執行緒間對鎖的爭用，進而減少阻塞，提高併發度。

記號②表示通過呼叫TransportContext的createServer方法建立傳輸服務端TransportServer的例項。在構造TransportServer的例項時，需要傳遞TransportContext、host、port、RpcHandler以及服務端載入程式TransportServerBootstrap的列表。

       有了對Spark內建RPC框架的基本架構的瞭解，現在正式介紹Spark的RPC框架所包含的各個元件：

• TransportContext：傳輸上下文，包含了用於建立傳輸服務端（TransportServer）和傳輸客戶端工廠（TransportClientFactory）的上下文資訊，並支援使用TransportChannelHandler設定Netty提供的SocketChannel的Pipeline的實現。
• TransportConf：傳輸上下文的配置資訊。
• RpcHandler：對呼叫傳輸客戶端（TransportClient）的sendRPC方法傳送的訊息進行處理的程式。
• MessageEncoder：在將訊息放入管道前，先對訊息內容進行編碼，防止管道另一端讀取時丟包和解析錯誤。
• MessageDecoder：對從管道中讀取的ByteBuf進行解析，防止丟包和解析錯誤；
• TransportFrameDecoder：對從管道中讀取的ByteBuf按照資料幀進行解析；
• RpcResponseCallback：RpcHandler對請求的訊息處理完畢後，進行回撥的介面。
• TransportClientFactory：建立傳輸客戶端（TransportClient）的傳輸客戶端工廠類。
• ClientPool：在兩個對等節點間維護的關於傳輸客戶端（TransportClient）的池子。ClientPool是TransportClientFactory的內部元件。
• TransportClient：RPC框架的客戶端，用於獲取預先協商好的流中的連續塊。TransportClient旨在允許有效傳輸大量資料，這些資料將被拆分成幾百KB到幾MB的塊。當TransportClient處理從流中獲取的獲取的塊時，實際的設定是在傳輸層之外完成的。sendRPC方法能夠在客戶端和服務端的同一水平線的通訊進行這些設定。
• TransportClientBootstrap：當服務端響應客戶端連線時在客戶端執行一次的載入程式。
• TransportRequestHandler：用於處理客戶端的請求並在寫完塊資料後返回的處理程式。
• TransportResponseHandler：用於處理服務端的響應，並且對發出請求的客戶端進行響應的處理程式。
• TransportChannelHandler：代理由TransportRequestHandler處理的請求和由TransportResponseHandler處理的響應，並加入傳輸層的處理。
• TransportServerBootstrap：當客戶端連線到服務端時在服務端執行一次的載入程式。
• TransportServer：RPC框架的服務端，提供高效的、低級別的流服務。

拓展知識：為什麼需要MessageEncoder和MessageDecoder？因為在基於流的傳輸裡（比如TCP/IP），接收到的資料首先會被儲存到一個socket接收緩衝裡。不幸的是，基於流的傳輸並不是一個數據包佇列，而是一個位元組佇列。即使你傳送了2個獨立的資料包，作業系統也不會作為2個訊息處理而僅僅認為是一連串的位元組。因此不能保證遠端寫入的資料會被準確地讀取。舉個例子，讓我們假設作業系統的TCP/TP協議棧已經接收了3個數據包：ABC、DEF、GHI。由於基於流傳輸的協議的這種統一的性質，在你的應用程式在讀取資料的時候有很高的可能性被分成下面的片段：AB、CDEFG、H、I。因此，接收方不管是客戶端還是服務端，都應該把接收到的資料整理成一個或者多個更有意義並且讓程式的邏輯更好理解的資料。

本文只是從整體上對Spark內建的RPC框架進行介紹，今後將分別介紹RPC框架的各個組成部分，他們是：

[1]  Akka是基於Actor併發程式設計模型實現的併發的分散式的框架。Akka是用Scala語言編寫的，它提供了Java和Scala兩種語言的API，減少開發人員對併發的細節處理，並保證分散式呼叫的最終一致性。在附錄B中有關於Akka的進一步介紹，感興趣的讀者不妨一讀。

[2]  Jetty 是一個開源的Servlet容器，它為基於Java的Web容器，例如JSP和Servlet提供執行環境。Jetty是使用Java語言編寫的，它的API以一組JAR包的形式釋出。開發人員可以將Jetty容器例項化成一個物件，可以迅速為一些獨立執行的Java應用提供網路和Web連線。在附錄C中有對Jetty的簡單介紹，感興趣的讀者可以選擇閱讀。

[3]  Netty是由Jboss提供的一個基於NIO的客戶、伺服器端程式設計框架，使用Netty 可以確保你快速、簡單的開發出一個網路應用，例如實現了某種協議的客戶，服務端應用。附錄G中有對Netty的簡單介紹，感興趣的讀者可以一讀。

關於《Spark核心設計的藝術 架構設計與實現》

經過近一年的準備，基於Spark2.1.0版本的《Spark核心設計的藝術 架構設計與實現》一書現已出版發行，圖書如圖：

紙質版售賣連結如下：