上一篇文章，簡單地用AKKA實現了RPC通訊，這篇文章在之前的基礎之上，進行了更多地關於RPC通訊的操作，包括傳一些相關資料，包括心跳機制等

具體功能如下圖：

程式碼結構：

WorkerInfo.scala程式碼

package cn.heres.rpc

/**
  * Created by vinsuan on 2017/6/15 0015.
  */
class WorkerInfo(val id: String,val memory:Int,val cores:Int) {
  //todo 上一次心跳
  var lastHeartBeatTime : Long = _
}

RemoteMessage.scala程式碼

package cn.heres.rpc

/**
  * Created by vinsuan on 2017/6/15 0015.
  */
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
 

Worker.scala程式碼：

package cn.heres.rpc

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

/**
  * Created by vinsuan on 2017/6/13 0013.
  */
class Worker(val masterHost : String ,val masterPost : Int,val memory: Int ,val cores :Int) extends Actor{

var master : ActorSelection = _
  val workerId = UUID.randomUUID().toString
  val HEARTBEAT_INTERVAL = 10000
//建立連線
  override def preStart(): Unit = {
     master = context.actorSelection(s"akka.tcp://
 
[email protected]$masterHost:$masterPost/user/Master")
    master ! RegisterWorker(workerId,memory,cores)
  }

  override def receive: Receive = {
    case RegisteredWorker(masterUrl) => {
      println(masterUrl)
      //匯入隱式轉換
      import context.dispatcher
      //啟動定時器傳送心跳
      context.system.scheduler.schedule(0 millis,HEARTBEAT_INTERVAL millis,self,SendHeartbeat)
    }
    case SendHeartbeat => {
      println("send heartbeat to master")
      master ! Heartbeat(workerId)
    }
  }

}
object Worker{
  def main(args: Array[String]) {

    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)
    //ActorSystem 老大，輔助建立和監控下面的Actor ，它是單例的
    val actorSystem = ActorSystem("workerSystem",config)
    //建立Actor
    val worker = actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")
    actorSystem.awaitTermination()
  }
}

Maser.scala程式碼：

package cn.heres.rpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

/**
  * Created by vinsuan on 2017/6/13 0013.
  */
class Master(val host:String,val port:Int) extends  Actor {
  //workerId -> workerInfo
  val idtoWorker = new scala.collection.mutable.HashMap[String, WorkerInfo]()
  //workerInfo
  val workers = new scala.collection.mutable.HashSet[WorkerInfo]()
  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, cores) => {
      //判斷一下，是不是已經註冊過
      if (!idtoWorker.contains(id)) {
        //把Worker的資訊封裝起來儲存到記憶體中
        val workerInfo = new WorkerInfo(id, memory, cores)
        idtoWorker(id) = workerInfo
        workers += workerInfo
        sender ! RegisteredWorker(s"akka.tcp://[email protected]$host:$port/user/Master")
      }
    }
    case Heartbeat(id) => {
      if (idtoWorker.contains(id)) {
        val workerInfo = idtoWorker(id)
        //報活
        val currentTime = System.currentTimeMillis()
        workerInfo.lastHeartBeatTime = currentTime

      }
    }
    case CheckTimeOutWorker => {
      val currentTime = System.currentTimeMillis()
      val toRemove = workers.filter(x => currentTime - x.lastHeartBeatTime > CHECK_INTERVAL)
      for (w <- toRemove) {
        workers -= w
        idtoWorker -= w.id
      }
      println("還有幾個活著的worker：" + workers.size)
    }
  }
}


object Master {
  def main(args: Array[String]) {

    val host = args(0)
    val port = args(1).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)
    //ActorSystem 老大，輔助建立和監控下面的Actor ，它是單例的
    val actorSystem = ActorSystem("MasterSystem", config)
    //建立Actor
    val master = actorSystem.actorOf(Props(new Master(host, port)), "Master")
    actorSystem.awaitTermination()
  }
}