[大資料]Scala 利用Akka 實現RPC框架
阿新 • • 發佈：2018-12-19
Scala 利用Akka 實現RPC框架
1 介紹
- 專案介紹：目前大多數的分散式架構底層通訊都是通過RPC實現的，RPC框架非常多，比如前我們學過的Hadoop專案的RPC通訊框架，但是Hadoop在設計之初就是為了執行長達數小時的批量而設計的，在某些極端的情況下，任務提交的延遲很高，所有Hadoop的RPC顯得有些笨重。 Spark 的RPC是通過Akka類庫實現的，Akka用Scala語言開發，基於Actor併發模型實現，Akka具有高可靠、高效能、可擴充套件等特點，使用Akka可以輕鬆實現分散式RPC功能。
- Akka程式設計模型介紹 Actor是Akka中最核心的概念，它是一個封裝了狀態和行為的物件，Actor之間可以通過交換訊息的方式進行通訊，每個Actor都有自己的收件箱（Mailbox）。通過Actor能夠簡化鎖及執行緒管理，可以非常容易地開發出正確地併發程式和並行系統，Actor具有如下特性：
- 1 提供了一種高階抽象，能夠簡化在併發（Concurrency）/並行（Parallelism）應用場景下的程式設計開發
- 2.提供了非同步非阻塞的、高效能的事件驅動程式設計模型
- 3.超級輕量級事件處理（每GB堆記憶體幾百萬Actor）
2 架構圖
- Master 啟動 等待Worker連線
- Worker啟動 向Master傳送註冊訊息
- Master 接受到註冊訊息 向Worker傳送已經註冊的訊息
- 每隔一段事件 Worker向Master 傳送心跳資訊
- Maste 在間隔事件定時清理心跳過期的Worker
3 重要類介紹
3.1 AkkaSystem
在Akka中，ActorSystem是一個重量級的結構，他需要分配多個執行緒，所以在實際應用中，ActorSystem通常是一個單例物件，我們可以使用這個ActorSystem建立很多Actor。
3.2 Actor
在Akka中，Actor負責通訊，在Actor中有一些重要的生命週期方法。
-
preStart()方法：該方法在Actor物件構造方法執行後執行，整個Actor生命週期中僅執行一次。
-
receive()方法：該方法在Actor的preStart方法執行完成後執行，用於接收訊息，會被反覆執行。
3.3 Master 及其伴生物件
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
import scala.collection.mutable
/**
* Created by 57871 on 2018/11/1. */
class Master(val host: String,val port: Int ) extends Actor{
val idToWorker = new mutable.HashMap[String,WorkerInfo]
// WorkerInfo
val workers = new mutable.HashSet[WorkerInfo]() //使用set刪除快, 也可用linkList //超時檢查的間隔
val CHECK_INTERVAL = 15000
override def preStart(): Unit = {
println("preStart invoked")
import context.dispatcher
context.system.scheduler.schedule(0 millis,CHECK_INTERVAL millis,self,CheckTimeOutworker)
}
override def receive: Receive = {
case RegisterWorker(id,memory,coreNum) => {
//判斷一下是不是註冊過了
if (!idToWorker.contains(id)) {
println(s"開始註冊 $id")
val worker = new WorkerInfo(id, memory, coreNum)
idToWorker(id) = worker
workers += worker
sender ! RegisteredWork(s"akka.tcp:// 3.2 Worker 及其伴生物件
import java.util.UUID
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
* Created by 57871 on 2018/11/2. */
class Worker (val masterHost: String, val masterPort: Int, memory:Int, coreNum: Int) extends Actor{
var master: ActorSelection = _
val workerId = UUID.randomUUID().toString
val HEART_INTERVAL = 10000 override def preStart(): Unit = {
//跟Master建立連線
master = context.actorSelection(s"akka.tcp://[email protected]$masterHost:$masterPort/user/Master")
println("開始建立連線")
master ! RegisterWorker(workerId,memory,coreNum)
}
override def receive: Receive = {
case RegisteredWork(masterUrl) =>{
println(masterUrl)
//啟動定時器傳送心跳資訊 import context.dispatcher
// 多長時間後執行 單位,多長時間執行一次 單位, 訊息的接受者(直接給master發不好, 先給自己傳送訊息, 以後可以做下判斷, 什麼情況下再發送訊息), 資訊 context.system.scheduler.schedule(0 millis,HEART_INTERVAL millis,self,SendHeartbeat)
}
case SendHeartbeat =>{
println("send heartbeat")
master ! Heartbeat(workerId)
}
}
}
object Worker {
def main(args: Array[String]): Unit = {
val host = args(0)
val port = args(1).toInt
val masterHost = args(2)
val masterPort = args(3).toInt
val memory = args(4).toInt
val cores = args(5).toInt
//準備配置
val configStr=
s"""
|akka.actor.provider = "akka.remote.RemoteActorRefProvider" |akka.remote.netty.tcp.hostname = "$host"
|akka.remote.netty.tcp.port = "$port" """.stripMargin
val config = ConfigFactory.parseString(configStr)
val actorSystem = ActorSystem("WorkerSystem",config)
actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")
actorSystem.registerOnTermination()
}
}
3.3 通訊資訊類
package cn.itcast.rpc
/**
* Created by root on 2016/5/13. */
trait RemoteMessage extends Serializable
//Worker -> Master case class RegisterWorker(id: String, memory: Int, cores: Int) extends RemoteMessage
case class Heartbeat(id: String) extends RemoteMessage
//Master -> Worker case class RegisteredWorker(masterUrl: String) extends RemoteMessage
//Worker -> self case object SendHeartbeat
// Master -> self case object CheckTimeOutWorker
3.4 WorkerInfo
package cn.itcast.rpc
/**
* Created by root on 2016/5/13. */
class WorkerInfo(val id: String, val memory: Int, val cores: Int) {
//TODO 上一次心跳
var lastHeartbeatTime : Long = _
}
4 效果實現
開始註冊 9fffc07c-cbec-4740-ad19-5d553bc7b42d
接受心跳
接受心跳
接受心跳
接受心跳
接受心跳
[WARN] [11/02/2018 11:56:13.513] [New I/O worker #5] [NettyTransport(akka://MasterSystem)] Remote connection to [/192.168.5.1:56987] failed with java.io.IOException: 遠端主機強迫關閉了一個現有的連線。
[WARN] [11/02/2018 11:56:13.514] [MasterSystem-akka.remote.default-remote-dispatcher-5] [akka.tcp://[email protected]:9999/system/endpointManager/reliableEndpointWriter-akka.tcp%3A%2F%2FWorkerSystem%40192.168.5.1%3A8888-1] Association with remote system [akka.tcp://[email protected]:8888] has failed, address is now gated for [5000] ms. Reason: [Disassociated]
去除掉失去心跳的worker 9fffc07c-cbec-4740-ad19-5d553bc7b42d