Spark RPC框架源碼分析（一）簡述

一. Spark rpc框架概述

Spark是最近幾年已經算是最為成功的大數據計算框架，那么這次我們就來介紹它內部的一個小點，Spark RPC框架。

在介紹之前，我們需要先說明什么是RPC，引用百度百科：

RPC（Remote Procedure Call）—遠程過程調用，它是一種通過網絡從遠程計算機程序上請求服務，而不需要了解底層網絡技術的協議。RPC協議假定某些傳輸協議的存在，如TCP或UDP，為通信程序之間攜帶信息數據。

Spark RPC可以說是Spark分布式集群的基礎，若是將Spark類比為一個人的話，Spark RPC無疑就是它的血液部分。而在Spark1.6之前，它的RPC部分還是用akka實現的，但之后底層就換成了netty來實現。為什么要這樣做呢？因為啊，這樣將Spark和Akka耦合在了一起，如果你系統本身就有使用到Akka，然后又想使用Spark的話，那兩個Akka框架版本不一致可怎么辦呀，這無疑是很讓人頭痛的。Spark團隊正是考慮到了這一點，所以將Akka替換成了netty。

這次我們就來看看Spark是如何讓它的血液流動起來的吧。有一位大神將Spark RPC中的RPC部分剝離出來，弄成一個新的可運行的　RPC　項目，這個項目本身就可以當作一個簡易的Akka來使用，地址在這Spark RPC。

雖然名字不一樣，但這個項目的類和內容基本和Spark Core中RPC部分的代碼和結構基本是一樣的，這樣我們就可以通過這個來學習Spark RPC框架。

PS：所用spark版本：spark2.1.0

二.Spark RPC中的 Hello world

我們程序員學東西最喜歡從一個Hello world開始，那么接下來我們就來演示如何下載並運行最簡單的Hello World例子吧。

首先，我使用的編譯器是IDEA，通過idea將github上的代碼clone下來。
可以看到項目目錄下有兩個模塊，

• kraps-rpc
• kraps-rpc-example

kraps-rpc存放的是Spark RPC的源代碼，而我們要做的即是運行 kraps-rpc-example中的示例代碼。

啟動PRC的話首先需要啟動Server端，開啟監聽服務，然后才能通過Client進行訪問。這里在HelloworldServer.scala中都已經幫我們寫好，不過在main方法中需要修改一下內容，就是將host改為本機地址。

  def main(args: Array[String]): Unit = {
//    val host = args(0)
    val host = "localhost"
    val config = RpcEnvServerConfig(new RpcConf(), "hello-server", host, 52345)
    val rpcEnv: RpcEnv = NettyRpcEnvFactory.create(config)
    val helloEndpoint: RpcEndpoint = new HelloEndpoint(rpcEnv)
    rpcEnv.setupEndpoint("hello-service", helloEndpoint)
    rpcEnv.awaitTermination()
  }

然后我們只需要右鍵該文件然后執行即可。

接下來我們就需要啟動Client端代碼，我們先到HelloworldClient文件中，這里面提供了同步和異步兩個方法可以運行。代碼同樣都已經寫好，通過修改注釋即可使用不同的方法運行。同樣是右鍵點擊該文件執行。

  def main(args: Array[String]): Unit = {
    //異步方法
    //asyncCall()
    //同步方法
    syncCall()
  }

異步方法中，ask會返回一個Future（注意這里的Future是scala中的Future，和java的是不一樣的）。並且在Future運行結果出來前，我們可以去做其他事情（異步的優勢所在）。scala中的Future和Java的Future有些不同，不過這可以先不去管，先當作Java里面的Future即可。

  def asyncCall() = {
    val rpcConf = new RpcConf()
    val config = RpcEnvClientConfig(rpcConf, "hello-client")
    val rpcEnv: RpcEnv = NettyRpcEnvFactory.create(config)
    val endPointRef: RpcEndpointRef = rpcEnv.setupEndpointRef(RpcAddress("localhost", 52345), "hello-service")
    val future: Future[String] = endPointRef.ask[String](SayHi("neo"))
    future.onComplete {
      case scala.util.Success(value) => println(s"Got the result = $value")
      case scala.util.Failure(e) => println(s"Got error: $e")
    }
    Await.result(future, Duration.apply("3s"))
    //在future結果運行出來前，會先打印這條語句。
    println("print me at first!")
    Thread.sleep(7)
  }

而同步方法是直接將結果返回，並且會阻塞，這個時間內你無法做其他事情，只能等待，直到結果返回。

  def syncCall() = {
    val rpcConf = new RpcConf()
    val config = RpcEnvClientConfig(rpcConf, "hello-client")
    val rpcEnv: RpcEnv = NettyRpcEnvFactory.create(config)
    val endPointRef: RpcEndpointRef = rpcEnv.setupEndpointRef(RpcAddress("localhost", 52345), "hello-service")
    val result = endPointRef.askWithRetry[String](SayBye("neo"))
    println(result)

  }

很簡單是吧，運行過例子后，我們就可以來了解一些Spark RPC運行過程中至關重要的兩個編程模型，以及在這其中使用到的一些主要的類。

三.Spark RPC中的兩個編程模型以及各個類

Spark RPC是使用了Actor模型和Reactor模型的混合模式，我們結合兩種模型分別說明Spark RPC中各個類的作用：

首先我們先來看Spark RPC的類圖。

是不是感覺很亂？沒事，我們來逐步剖析各個類。

為了更加清楚了說明各個類的關系，我們要先知道兩個模型，分別是Actor模型和Reactor模型，我們將從這兩個模型的角度來拆解各個類的關系。

Actor模型

其實之前也有寫過一篇介紹Actor模型的文章，感興趣的同學可以點擊這里查看Actor模型淺析。

其實Actor主要就是這副圖的內容：

在Spark RPC中有幾個類分別與Actor模型中的各個角色對應，對應如下，左邊的是Spark RPC中的類，右邊的是Actor模型中的角色：

RpcEndpoint => Actor

RpcEndpointRef => ActorRef

RpcEnv => ActorSystem

我們逐個來看：

RpcEnv --RPC Environment

RPC Environment 是 RpcEndpoint 的運行環境。它管理 RpcEndpoint 的整個生命周期：

1. 通過名字或 URI 注冊 RpcEndpoint。
2. 對到底的消息進行路由，決定分發給哪個 RpcEndpoint。
3. 停止 RpcEndpoint。

RPC Environment在akka已經被移除的2.0后面版本中，RPC Environment的實現類是NettyRpcEnv。通常是由NettyRpcEnvFactory.create創建。

RpcEndpoint

RpcEndpoint能通過callbacks接收消息。通常需要我們自己寫一個類繼承RpcEndpoint。編寫自己的接收信息和返回信息規則。

RpcEndpoint的生命周期被RPC Environment管理。其生命周期包括，onStart,receive和onStop。

它是作為服務端，比如上面例子中的HelloworldServer就是一個RpcEndpoint。

RpcEndpointRef

RpcEndpointRef是RpcEndpoint在RPC Environment中的一個引用。

它包含一個地址（即Spark URL）和名字。RpcEndpointRef作為客戶端向服務端發送請求並接收返回信息，通常可以選擇使用同步或異步的方式進行發送。

Reactor模型

Spark RPC采用Actor模型和Reactor模型混合的結構，上面已經介紹了Actor，那么現在我們就來介紹Reactor模型，同樣，我們可以從一張圖來看Reactor的架構。

使用Reactor模型，由底層netty創建的EventLoop做I/O多路復用，這里使用Multiple Reactors這種形式，如上圖所示，從netty的角度而言，Main Reactor和Sub Reactor對應BossGroup和WorkerGroup的概念，前者負責監聽TCP連接、建立和斷開，后者負責真正的I/O讀寫。

而圖中的ThreadPool就是的Dispatcher中的線程池，它來解耦開來耗時的業務邏輯和I/O操作，這樣就可以更scalabe，只需要少數的線程就可以處理成千上萬的連接，這種思想是標准的分治策略，offload非I/O操作到另外的線程池。

Dispatcher

Dispatcher的主要作用是保存注冊的RpcEndpoint、分發相應的Message到RpcEndPoint中進行處理。Dispatcher即是上圖中ThreadPool的角色。它同時也維系一個threadpool，用來處理每次接受到的InboxMessage。而這里處理InboxMessage是通過inbox實現的。

Inbox

Inbox其實屬於Actor模型，是Actor中的信箱，不過它和Dispatcher聯系緊密所以放這邊。

InboxMessage有多個實現它的類，比如OneWayMessage，RpcMessage，等等。Dispatcher會將接收到的InboxMessage分發到對應RpcEndpoint的Inbox中，然后Inbox便會處理這個InboxMessage。

OK，這次就先介紹到這里，下次我們從代碼的角度來看Spark RPC的運行機制

如果覺得對你有幫助，不妨關注一波吧~~

參考資料：https://zhuanlan.zhihu.com/p/28893155

---

推薦閱讀：
從分治算法到 MapReduce
Actor並發編程模型淺析
大數據存儲的進化史 --從 RAID 到 Hadoop Hdfs
一個故事告訴你什么才是好的程序員