Scala使用Akka模拟RPC机制代码2

//对象要序列化才能通过网络传输   这个地方没有大括号....这有这个extends声明
trait RemoteMessage extends Serializable

//Worker ->(发送给) Master  Worker给Master节点发送注册消息
case class RegisterWorker(id:String, memory:Int, scores:Int ) extends RemoteMessage

//接收已经在Master注册过的Worker定时发送来的心跳  Woker->Master
case class Heartbeat(id:String) extends RemoteMessage

//Master -> Worker   Master反馈给Worker已经注册的Worker有哪些
case class RegisteredWorker(masterUrl:String) extends RemoteMessage

//Worker -> self  //Worker先向自己发送一下心跳,这个地方可以做一下判断,根据实际情况,定义什么时候向Master发送消息
//在这个样例类中发送给Master的心跳  case class Heartbeat
case object SendHeartbeat

//Master -> self  Master自检查看Worker的状态
case object CheckTimeOutWorker

class WorkerInfo(val id:String,val memory:Int,val scores:Int) {
  //记录上一次心跳
  var lastHeartbeatTime : Long = _
}

import java.util.UUID
import com.typesafe.config.ConfigFactory
import akka.actor.Actor
import akka.actor.ActorSystem
import akka.actor.Props
import akka.actor.ActorSelection
import scala.concurrent.duration._  //如果不引入这个包 在 context.system.scheduler.schedule 中的mills 地方会报错

//使用Akka框架来实现RCP不同进程之间的方法调用   导入Actor是导入的akka包下的
//在Worker类上定义一个主构造器
class Worker (val masterHost:String ,val masterPort:Int ,val memory:Int ,val cores: Int) extends Actor {
  var master : ActorSelection = _
  //记录Worker从节点的id
  val workerId = UUID.randomUUID().toString()
  //记录Worker从节点的发送心跳的时间间隔
  val HEART_INTERVAL = 10000
  
  //preStart()方法在构造器之后,在receive之前执行  actorOf()方法执行 就会执行生命周期方法
  override def preStart() : Unit={
    //跟Master建立连接
    master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")
    //向Master发送注册消息
    master ! RegisterWorker(workerId,memory,cores)
  }
  
  override def receive: Receive = {
    //Master对注册的Worker存储注册之后需要向Worker发送消息 来说明该Woker注册成功
    //此处是Worker节点接收到Master节点的反馈消息
    case RegisteredWorker(masterUrl) => {
      println(masterUrl)
      //启动定时器发送心跳
      import context.dispatcher
      //多长时间后执行 单位 ,多长时间执行一次 单位  ,消息的接受者(直接给master发不好,先给自己发送消息,以后可以做下判断,什么情况下再发送消息), 消息
      context.system.scheduler.schedule(0 millis, HEART_INTERVAL millis, self, SendHeartbeat)
    }
    //Worker先向自己self发送一下心跳,这个地方可以做一下判断,根据实际情况,定义什么时候向Master发送消息
    case SendHeartbeat =>{
      println("send heartbeat to Master")
      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
    //为了通过ActorSystem老大来获得Actor  准备配置
    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的老大,辅助创建和监控下面的Actor, 该老大是单例的.
    //得到该老大的对象需要一个名字和配置对象config
    val actorSystem = ActorSystem("WorkerSystem",config) 
    //actorOf方法执行,就会执行生命周期方法
    actorSystem.actorOf(Props(new Worker(masterHost, masterPort, memory, cores)),"Worker")
    actorSystem.awaitTermination()
  }
}

import com.typesafe.config.ConfigFactory

import akka.actor.Actor
import akka.actor.ActorSystem
import akka.actor.Props
import scala.collection.mutable
import scala.concurrent.duration._

class Master(val host:String,val port:Int) extends Actor{
  
  //定义一个HashMap用来存储注册Worker的信息
  val idToWorkerHashMap = new mutable.HashMap[String,WorkerInfo]()
  //定义一个HashSet用于存储 注册的Worker对象 WorkerInfo
  val workersSet = new mutable.HashSet[WorkerInfo]() //使用set删除快, 也可用linkList
  //超时检查的间隔  这个时间间隔一定要大于Worker向Master发送心跳的时间间隔.
  val CHECK_INTERVAL = 15000
  
  override def preStart() :Unit = {
    println("preStart invoked")
    //导入隐式转换
    //使用timer太low了,可以使用akka的定时器,需要导入这个包
    import context.dispatcher 
    context.system.scheduler.schedule(0 millis, CHECK_INTERVAL millis, self, CheckTimeOutWorker)
  }
  //接收Worker发送来的消息  可以有注册RegisterWorker 心跳Heartbeat
  override def receive :Receive = {
    case RegisterWorker(id,memory,cores) =>{
      //判断该Worker是否已经注册过 
      if(!idToWorkerHashMap.contains(id)){
        //把Worker的信息封装起来保存到内存中
        val workerInfo = new WorkerInfo(id,memory,cores)
        //以id为键  worker对象为值 添加到该idToWorker的HashMap对象中
        idToWorkerHashMap(id) = workerInfo 
        //把workerInfo对象放到存储Woker的HashSet中
        workersSet += workerInfo
        //对注册的Worker存储注册之后需要向Worker发送消息 来说明该Woker注册成功
        sender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")//通知worker注册
      }
    }
    //接收已经在Master注册过的Worker定时发送来的心跳 
    case Heartbeat(id) =>{
      if(idToWorkerHashMap.contains(id)){
        val workerInfo = idToWorkerHashMap(id)
        //报活  把该Worker给Master心跳的时间记录下来
        val currentTime = System.currentTimeMillis()
        workerInfo.lastHeartbeatTime = currentTime
      }
    }
    //Master节点主动检测在Master自己这里注册Worker节点的存活状态
    case CheckTimeOutWorker =>{
      val currentTime = System.currentTimeMillis();
      //如果当前时间和最近一次心跳时间的差值大于检测的时间间隔 就要干掉这个Worker
      val toRemove = workersSet.filter(x => currentTime - x.lastHeartbeatTime > CHECK_INTERVAL)
      for(w <- toRemove){
        workersSet -= w
        idToWorkerHashMap -= w.id
      }
      println(workersSet.size)      
    }
  }
}

object Master{
  def main(args: Array[String]): Unit = {
    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
    //ConfigFactory 既有parseString 也有parseFile 
    val config = ConfigFactory.parseString(configStr)
    //ActorSystem是所有Actor的老大,辅助创建和监控下面的所有Actor,它是单例的
    val actorSystem = ActorSystem("MasterSystem",config)
    val master = actorSystem.actorOf(Props(new Master(host,port)),"Master")
    actorSystem.awaitTermination()
  }
}