2. 程式人生 > >Spark RPC 框架原始碼分析（二）執行時序

Spark RPC 框架原始碼分析（二）執行時序

阿新 • • 發佈：2019-01-14

前情提要：

一. Spark RPC 概述概述

上一篇我們已經說明了 Spark RPC 框架的一個簡單例子，以及一些基本概念的說明。這一篇我們主要講述其執行時序，從而揭露 Spark RPC 框架的執行原理。我們將分為兩部分，分別從服務端和客戶端來看。

所用 spark 版本：spark 2.1.0

二. Spark RPC 服務端

我們以上一篇 HelloworldServer 為線索，深入到 Spark RPC 框架來看看啟動一個服務時都做了些什麼。

HelloworldServer{
  ......
  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()
  }
  ......
}

這段程式碼中有兩個主要流程，我們分別來說

2.1 Spark RPC 服務端 NettyRpcEnvFactory.create(config)

首先是下面這條程式碼的執行流程：

val rpcEnv: RpcEnv = NettyRpcEnvFactory.create(config)

其實就是通過 NettyRpcEnvFactory 創建出一個 RPC Environment ，其具體類是 NettyRpcEnv 。

我們再來看看建立過程中會發生什麼。

object NettyRpcEnvFactory extends RpcEnvFactory {
    ......
    def create(config: RpcEnvConfig): RpcEnv = {
        val conf = config.conf
    
        // Use JavaSerializerInstance in multiple threads is safe. However, if we plan to support
        // KryoSerializer in future, we have to use ThreadLocal to store SerializerInstance
        val javaSerializerInstance =
        new JavaSerializer(conf).newInstance().asInstanceOf[JavaSerializerInstance]
        //根據配置以及地址，new 一個 NettyRpcEnv ，
        val nettyEnv =
        new NettyRpcEnv(conf, javaSerializerInstance, config.bindAddress)
        //如果是服務端建立的，那麼會啟動服務。服務端和客戶端都會通過這個方法建立一個 NettyRpcEnv ，但區別就在這裡了。
        if (!config.clientMode) {
        val startNettyRpcEnv: Int => (NettyRpcEnv, Int) = { actualPort =>
            //啟動服務的方法,下一步就是呼叫這個方法了
            nettyEnv.startServer(config.bindAddress, actualPort)
            (nettyEnv, nettyEnv.address.port)
        }
        try {
            Utils.startServiceOnPort(config.port, startNettyRpcEnv, conf, config.name)._1
        } catch {
            case NonFatal(e) =>
            nettyEnv.shutdown()
            throw e
        }
        }
        nettyEnv
    }
    ......
}

還沒完，如果是服務端呼叫這段程式碼，那麼主要的功能是建立 RPCEnv ，即 NettyRpcEnv（客戶端在後面說） 。以及通過下面這行程式碼，

nettyEnv.startServer(config.bindAddress, actualPort)

去呼叫相應的方法啟動服務端的服務。下面進入到這個方法中去看看。

class NettyRpcEnv(
                   val conf: RpcConf,
                   javaSerializerInstance: JavaSerializerInstance,
                   host: String) extends RpcEnv(conf) {
  ......
  def startServer(bindAddress: String, port: Int): Unit = {
    // here disable security
    val bootstraps: java.util.List[TransportServerBootstrap] = java.util.Collections.emptyList()
    //TransportContext 屬於 spark.network 中的部分，負責 RPC 訊息在網路中的傳輸
    server = transportContext.createServer(bindAddress, port, bootstraps)
    //在每個 RpcEndpoint 註冊的時候都會註冊一個預設的 RpcEndpointVerifier，它的作用是客戶端呼叫的時候先用它來詢問 Endpoint 是否存在。
    dispatcher.registerRpcEndpoint(
      RpcEndpointVerifier.NAME, new RpcEndpointVerifier(this, dispatcher))
  }
  ......
}

執行完畢之後這個 create 方法就結束。這個流程主要就是開啟一些服務，然後返回一個新的 NettyRpcEnv 。

2.2 Spark RPC 服務端 rpcEnv.setupEndpoint("hello-service", helloEndpoint)

這條程式碼會去呼叫 NettyRpcEnv 中相應的方法

class NettyRpcEnv(
                   val conf: RpcConf,
                   javaSerializerInstance: JavaSerializerInstance,
                   host: String) extends RpcEnv(conf) {
  ......
  override def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef = {
    dispatcher.registerRpcEndpoint(name, endpoint)
  }
  ......
}

我們看到，這個方法主要是呼叫 dispatcher 進行註冊的。dispatcher 的功能上一節已經說了，

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

，這裡我們就說一說 dispatcher 的流程。

dispatcher

dispatcher 在 NettyRpcEnv 被建立的時候創建出來。

class NettyRpcEnv(
                   val conf: RpcConf,
                   javaSerializerInstance: JavaSerializerInstance,
                   host: String) extends RpcEnv(conf) {
    ......
    //初始化時建立 dispatcher
    private val dispatcher: Dispatcher = new Dispatcher(this)
    ......
}

dispatcher 類被建立的時候也有幾個屬性需要注意：

private[netty] class Dispatcher(nettyEnv: NettyRpcEnv) {
    ......
    //每個 RpcEndpoint 其實都會被整合成一個 EndpointData 。並且每個 RpcEndpoint 都會有一個 inbox。
    private class EndpointData(
                                val name: String,
                                val endpoint: RpcEndpoint,
                                val ref: NettyRpcEndpointRef) {
        val inbox = new Inbox(ref, endpoint)
    }
    
    //一個阻塞佇列，當有 RpcEndpoint 相關請求（InboxMessage）的時候，就會將請求塞到這個佇列中，然後被執行緒池處理。
    private val receivers = new LinkedBlockingQueue[EndpointData]
    
    //初始化便創建出來的執行緒池，當上面的 receivers 佇列中沒內容時，會阻塞。當有 RpcEndpoint 相關請求（即 InboxMessage ）的時候就會立刻執行。
    //這裡處理 InboxMessage 本質上是呼叫相應 RpcEndpoint 的 inbox 去處理。
    private val threadpool: ThreadPoolExecutor = {
        val numThreads = nettyEnv.conf.getInt("spark.rpc.netty.dispatcher.numThreads",
        math.max(2, Runtime.getRuntime.availableProcessors()))
        val pool = ThreadUtils.newDaemonFixedThreadPool(numThreads, "dispatcher-event-loop")
        for (i <- 0 until numThreads) {
            pool.execute(new MessageLoop)
        }
        pool
    }
    ......
}

瞭解一些 Dispatcher 的邏輯流程後，我們來正式看看 Dispatcher 的 registerRpcEndpoint 方法。

顧名思義，這個方法就是將 RpcEndpoint 註冊到 Dispatcher 中去。當有 Message 到來的時候，便會分發 Message 到相應的 RpcEndPoint 中進行處理。

private[netty] class Dispatcher(nettyEnv: NettyRpcEnv) {
  ......
  def registerRpcEndpoint(name: String, endpoint: RpcEndpoint): NettyRpcEndpointRef = {
    val addr = RpcEndpointAddress(nettyEnv.address, name)
    //註冊 RpcEndpoint 時需要的是 上面的 EndpointData ，其中就包含 endpointRef ，這個主要是供客戶端使用的。
    val endpointRef = new NettyRpcEndpointRef(nettyEnv.conf, addr, nettyEnv)
    //多執行緒環境下，註冊一個 RpcEndpoint 需要判斷現在是否處於 stop 狀態。
    synchronized {
      if (stopped) {
        throw new IllegalStateException("RpcEnv has been stopped")
      }
      //新建 EndpointData 並存儲到一個 ConcurrentMap 中。
      if (endpoints.putIfAbsent(name, new EndpointData(name, endpoint, endpointRef)) != null) {
        throw new IllegalArgumentException(s"There is already an RpcEndpoint called $name")
      }
      val data = endpoints.get(name)
      endpointRefs.put(data.endpoint, data.ref)
      //將 這個 EndpointData 加入到 receivers 佇列中，此時 dispatcher 中的 threadpool 會去處理這個加進來的 EndpointData 
      //處理過程是呼叫它的 inbox 的 process()方法。然後 inbox 會等待訊息到來。
      receivers.offer(data) // for the OnStart message
    }
    endpointRef
  }
  ......
}

Spark RPC 服務端邏輯小結：我們說明了 Spark RPC 服務端啟動的邏輯流程，分為兩個部分，第一個是 Spark RPC env ，即 NettyRpcEnv 的建立過程，第二個則是 RpcEndpoint 註冊到 dispatcher 的流程。
1. NettyRpcEnvFactory 建立 NettyRpcEnv

  • 根據地址建立 NettyRpcEnv。
  • NettyRpcEnv 開始啟動服務，包括 TransportContext 根據地址開啟監聽服務，向 Dispacther 註冊一個 RpcEndpointVerifier 等待。

2. Dispatcher 註冊 RpcEndpoint

  • Dispatcher 初始化時便建立一個執行緒池並阻塞等待 receivers 佇列中加入新的 EndpointData
  • 一旦新加入 EndpointData 便會呼叫該 EndpointData 的 inbox 去處理訊息。比如 OnStart 訊息，或是 RPCMessage 等等。

三.Spark RPC 客戶端

依舊是以上一節 Spark RPC 客戶端 HelloWorld 的為線索，我們來逐層深入 Spark RPC 客戶端 HelloworldClient 的 asyncCall() 方法。

object HelloworldClient {
  ......
  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("30s"))
    rpcEnv.shutdown()
  }
  ......
}


建立 Spark RPC 客戶端 Env（即 NettyRpcEnvFactory ） 部分和 Spark RPC 服務端是一樣的，只是不會開啟監聽服務，這裡就不詳細展開。

我們從這一句開始看，這也是 Spark RPC 客戶端和服務端區別的地方所在。

val endPointRef: RpcEndpointRef = rpcEnv.setupEndpointRef(RpcAddress("localhost", 52345), "hello-service")

setupEndpointRef()

上面的的 setupEndpointRef 最終會去呼叫下面 setupEndpointRef() 這個方法，這個方法中又進行一次跳轉，跳轉去 setupEndpointRefByURI 這個方法中 。需要注意的是這兩個方法都是 RpcEnv 裡面的，而 RpcEnv 是抽象類，它裡面只實現部分方法，而 NettyRpcEnv 繼承了它，實現了全部方法。

abstract class RpcEnv(conf: RpcConf) {
  ......
  def setupEndpointRef(address: RpcAddress, endpointName: String): RpcEndpointRef = {
    //會跳轉去呼叫下面的方法
    setupEndpointRefByURI(RpcEndpointAddress(address, endpointName).toString)
  }
  
  def setupEndpointRefByURI(uri: String): RpcEndpointRef = {
    //其中 asyncSetupEndpointRefByURI() 返回的是 Future[RpcEndpointRef]。 這裡就是阻塞，等待返回一個 RpcEndpointRef。
    // defaultLookupTimeout.awaitResult 底層呼叫 Await.result 阻塞 直到結果返回或返回異常
    defaultLookupTimeout.awaitResult(asyncSetupEndpointRefByURI(uri))
  }
  ......
}

這裡最主要的程式碼其實就一句，

defaultLookupTimeout.awaitResult(asyncSetupEndpointRefByURI(uri))

這一段可以分為兩部分， 第一部分的 defaultLookupTimeout.awaitResult 其實底層是呼叫 Await.result 阻塞等待一個非同步操作，直到結果返回。

而asyncSetupEndpointRefByURI(uri) 則是根據給定的 uri 去返回一個 RpcEndpointRef ，它是在 NettyRpcEnv 中實現的：

class NettyRpcEnv(
                   val conf: RpcConf,
                   javaSerializerInstance: JavaSerializerInstance,
                   host: String) extends RpcEnv(conf) {
  ......
  def asyncSetupEndpointRefByURI(uri: String): Future[RpcEndpointRef] = {
    //獲取地址
    val addr = RpcEndpointAddress(uri)
    //根據地址等資訊新建一個 NettyRpcEndpointRef 。
    val RpcendpointRef = new NettyRpcEndpointRef(conf, addr, this) 
    //每個新建的 RpcendpointRef 都有先有一個對應的verifier 去檢查服務端存不存在對應的 Rpcendpoint 。  
    val verifier = new NettyRpcEndpointRef(
      conf, RpcEndpointAddress(addr.rpcAddress, RpcEndpointVerifier.NAME), this)
    //向服務端傳送請求判斷是否存在對應的 Rpcendpoint。
    verifier.ask[Boolean](RpcEndpointVerifier.createCheckExistence(endpointRef.name)).flatMap { find =>
      if (find) {
        Future.successful(endpointRef)
      } else {
        Future.failed(new RpcEndpointNotFoundException(uri))
      }
    }(ThreadUtils.sameThread)
  }
  ......
}

asyncSetupEndpointRefByURI() 這個方法實現兩個功能，第一個就是新建一個 RpcEndpointRef 。第二個是新建一個 verifier ，這個 verifier 的作用就是先給服務端傳送一個請求判斷是否存在 RpcEndpointRef 對應的 RpcEndpoint 。

這段程式碼中最重要的就是 verifiter.ask[Boolean]（...） 了。如果有找到之後就會呼叫 Future.successful 這個方法，反之則會 通過 Future.failed 丟擲一個異常。

ask 可以算是比較核心的一個方法，我們可以到 ask 方法中去看看。

class NettyRpcEnv{
    ......
    private[netty] def ask[T: ClassTag](message: RequestMessage, timeout: RpcTimeout): Future[T] = {
      val promise = Promise[Any]()
      val remoteAddr = message.receiver.address
      //
      def onFailure(e: Throwable): Unit = {
  //      println("555");
        if (!promise.tryFailure(e)) {
          log.warn(s"Ignored failure: $e")
        }
      }
  
      def onSuccess(reply: Any): Unit = reply match {
        case RpcFailure(e) => onFailure(e)
        case rpcReply =>
          println("666");
          if (!promise.trySuccess(rpcReply)) {
            log.warn(s"Ignored message: $reply")
          }
      }
  
      try {
        if (remoteAddr == address) {
          val p = Promise[Any]()
          p.future.onComplete {
            case Success(response) => onSuccess(response)
            case Failure(e) => onFailure(e)
          }(ThreadUtils.sameThread)
          dispatcher.postLocalMessage(message, p)
        } else {
          //跳轉到這裡執行
          //封裝一個 RpcOutboxMessage ，同時 onSuccess 方法也是在這裡註冊的。
          val rpcMessage = RpcOutboxMessage(serialize(message),
            onFailure,
            (client, response) => onSuccess(deserialize[Any](client, response)))
          postToOutbox(message.receiver, rpcMessage)
          promise.future.onFailure {
            case _: TimeoutException =>  println("111");rpcMessage.onTimeout()
  //          case _ => println("222");
          }(ThreadUtils.sameThread)
        }
        
        val timeoutCancelable = timeoutScheduler.schedule(new Runnable {
          override def run(): Unit = {
  //          println("333");
            onFailure(new TimeoutException(s"Cannot receive any reply in ${timeout.duration}"))
          }
        }, timeout.duration.toNanos, TimeUnit.NANOSECONDS)
        //promise 對應的 future onComplete時會去呼叫，但當 successful 的時候，上面的 run 並不會被呼叫。
        promise.future.onComplete { v =>
  //        println("4444");
          timeoutCancelable.cancel(true)
        }(ThreadUtils.sameThread)
  
      } catch {
        case NonFatal(e) =>
          onFailure(e)
      }
  
      promise.future.mapTo[T].recover(timeout.addMessageIfTimeout)(ThreadUtils.sameThread)
    }
    ......
}

這裡涉及到使用一些 scala 多執行緒的高階用法，包括 Promise 和 Future。如果想要對這些有更加深入的瞭解，可以參考這篇文章。

這個函式的作用從名字中就可以看得出，其實就是將要 傳送的訊息封裝成一個 RpcOutboxMessage ，然後交給 OutBox 去傳送，OutBox 和前面所說的 InBox 對應，對應 Actor 模型中的 MailBox（信箱）。用於傳送和接收訊息。

其中使用到了 Future 和 Promise 進行非同步併發以及錯誤處理，比如當傳送時間超時的時候 Promise 就會返回一個 TimeoutException ，而我們就可以設定自己的 onFailure 函式去處理這些異常。

OK，註冊完 RpcEndpointRef 後我們便可以用它來向服務端傳送訊息了，而其實 RpcEndpointRef 傳送訊息還是呼叫 ask 方法，就是上面的那個 ask 方法。上面也有介紹，本質上就是通過 OutBox 進行處理。

我們來梳理一下客戶端的傳送流程。

客戶端邏輯小結：客戶端和服務端比較類似，都是需要建立一個 NettyRpcEnv 。不同的是接下來客戶端建立的是 RpcEndpointRef ，並用之向服務端對應的 RpcEndpoint 傳送訊息。

1. NettyRpcEnvFactory 建立 NettyRpcEnv

  • 根據地址建立 NettyRpcEnv。 根據地址開啟監聽服務，向 Dispacther 註冊一個 RpcEndpointVerifier 等待。

2. 建立 RpcEndpointRef

  • 建立一個新的 RpcEndpointRef
  • 建立對應的 verifier ，使用 verifier 向服務端傳送請求，判斷對應的 RpcEndpoint 是否存在。若存在，返回該 RpcEndpointRef ，否則丟擲異常。

3. RpcEndpointRef 使用同步或者非同步的方式傳送請求。

OK，以上就是 Spark RPC 時序的原始碼分析。下一篇會將一個實際的例子，Spark 的心跳機制和程式碼。喜歡的話就關注一波吧

推薦閱讀 ：
從分治演算法到 MapReduce
Actor併發程式設計模型淺析
大資料儲存的進化史 --從 RAID 到 Hadoop Hdfs
一個故事告訴你什麼才是好的程式設計師

相關推薦

Spark RPC 框架原始碼分析（二）執行時序

前情提要： Spark RPC 框架原始碼分析（一）簡述 一. Spark RPC 概述概述 上一篇我們已經說明了 Spark RPC 框架的一個簡單例子，以及一些基本概念的說明。這一篇我們主要講述其執行時序，從而揭露 Spark RPC 框架的執行原理。我們將分為兩部分，分別從服務端和客戶端

Openstack原始碼分析 Neutron原始碼分析（二）-------------rpc篇

上一篇分析了neutron wsgi應用的原始碼，這一篇分析另外一部分核心功能，rpc篇，同時分析一下neutron採用的併發模型。 還是上篇的程式碼，啟動完wsgi後，啟動rpc_workers。 neutron/server/wsgi_eventlet.py: def

Flume NG原始碼分析（二）支援執行時動態修改配置的配置模組

在上一篇中講了Flume NG配置模組基本的介面的類，PropertiesConfigurationProvider提供了基於properties配置檔案的靜態配置的能力，這篇細說一下PollingPropertiesFileConfigurationProvider提供的執行時動態修改配置並生效的

GCC原始碼分析（二）——前端

原文連結：http://blog.csdn.net/sonicling/article/details/6706152 　　從這一篇開始，我們將從原始碼的角度來分析GCC如何完成對C語言原始檔的處理。GCC的內部構架在GCC Internals（搜“gccint.pdf”，或者見[

Glide原始碼分析（二）——從用法來看之load&into方法

上一篇，我們分析了with方法，文章連結： https://blog.csdn.net/qq_36391075/article/details/82833260 在with方法中，進行了Glide的初始化，建立了RequesManger，並且綁定了生命週期，最終返回了一個Reques

YOLOv2原始碼分析（二）

文章全部YOLOv2原始碼分析 接著上一講沒有講完的make_convolutional_layer函式 0x01 make_convolutional_layer //make_convolutional_laye

zigbee 之ZStack-2.5.1a原始碼分析（二） 無線接收控制LED

本文描述ZStack-2.5.1a 模板及無線接收移植相關內容。 main HAL_BOARD_INIT // HAL_TURN_OFF_LED1 InitBoard HalDriverInit HalAdcInit

兄弟連區塊鏈入門教程eth原始碼分析p2p-udp.go原始碼分析（二）

ping方法與pending的處理，之前談到了pending是等待一個reply。 這裡通過程式碼來分析是如何實現等待reply的。pending方法把pending結構體傳送給addpending. 然後等待訊息的處理和接收。 // ping sends a ping message to the giv

Spring原始碼分析（二）（IoC容器的實現）（1）

    Ioc（Inversion of Control）——“控制反轉”，不是什麼技術，而是一種設計思想。在Java開發中，Ioc意味著將你設計好的物件交給容器控制，而不是傳統的在你的物件內部直接控制。理解好Ioc的關鍵是要明確“誰控制誰，控制什麼，為何是反轉（有

tornado原始碼分析（二）之iostream

在事件驅動模型中，所有任務都是以某個事件的回撥函式的方式新增至事件迴圈中的，如：HTTPServer要從socket中讀取客戶端傳送的request訊息，就必須將該socket新增至ioloop中，並設定回掉函式，在回掉函式中從socket中讀取資料，並且檢查request訊息是否全部接收到了，如果

Cat原始碼分析（二）：Server端

初始化 服務端消費客戶端發來的訊息進行分析和展示，所以這個的初始化指的是CatHomeModule的初始化 CatHomeModule依賴TcpSocketReceiver和MessageConsumer，前者用來接收客戶端傳送的訊息，後者用來消費訊息。 TcpSocket

laravel框架原始碼分析（一）自動載入

一、前言 　　使用php已有好幾年，laravel的使用也是有好長時間，但是一直對於框架原始碼的理解不深，原因很多，歸根到底還是php基礎不紮實，所以原始碼看起來也比較吃力。最近有時間，所以開啟第5、6遍的框架原始碼探索之旅，前面幾次都是看了一些就放棄，希望這次能夠看完。每一次看原始碼都會有新的收穫，因為框

subsampling-scale-image-view載入長圖原始碼分析（二）

subsampling-scale-image-view原始碼分析概要分析總結 概要 subsampling-scale-image-view是一個支援部分載入大圖長圖的圖片庫，並且還支援縮放，在subsampling-scale-image-view載入長圖原

Spring component-scan原始碼分析（二） -- @Configuration註解處理

上篇文章Spring component-scan原始碼分析（一） – XML解析分析了Spring解析<context:component-scan …/>標籤時，把掃描到的合適的類封裝成BeanDefinition加入Sping容器中，本篇分析S

Spring原始碼分析（二）（IoC容器的實現）（3）

BeanDefinition的載入和解析     這個載入過程，相當於把定義的BeanDefinition在IoC容器中轉化成一個Spring內部表示的資料結構的過程。IoC容器對Bean的管理和依賴注入功能的實現，是通過對其持有的BeanDefinition進

Spring原始碼分析（二）（IoC容器的實現）（2）

IoC容器的初始化過程     簡單來說IoC容器的初始化是由refresh()方法啟動的，這個方法標誌著IoC容器的正式啟動。這個啟動包括BeanDefinition的Resouce定位、載入和註冊三個基本過程。     第一

groupcache 原始碼分析（二）-- LRU

lru部分的程式碼在lru/lru.go檔案中，它主要是封裝了一系列lru演算法相關的介面，供groupcahe進行快取置換相關的呼叫。 它主要封裝了下面幾個介面： // 建立一個Cache func New(maxEntries int) *Cache /

Spark2.3.2原始碼解析： 7. SparkContext原始碼分析（二） ：TaskScheduler

    程式碼部分：   啟動指令碼 --name spark-test --class WordCount --master yarn --deploy-mode cluster /A/spark-test.jar /

jieba原始碼分析（二）

0、寫在前面 在jieba原始碼分析（一）裡面已經jieba分詞的一部分進行了分析，本文主要解決分詞的另一塊：未登陸詞，也就是我們常說的新詞。對於這些新詞，我們前面所說的字首詞典中是不存在的，那麼之前的分詞方法自然就不能適用了。為了解決這一問題，jieba使用了隱馬爾科夫（

2.gson-plugin基礎原始碼分析（二）

二、Gson解析核心類 1.ArrayTypeAdapter.JAVA 用於解析陣列型別的資料 public Object read(JsonReader in) throws IOException { if(in.peek() == JsonT