Akka Actor模拟实现YARN

1.需求：模拟实现YARN（具有注册和心跳的功能）

2.实现思路图

代码实现：
pom.xml

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.7</maven.compiler.source>
    <maven.compiler.target>1.7</maven.compiler.target>
      <scala.version>2.11.8</scala.version>
      <scala.actors.version>2.11.8</scala.actors.version>
      <akka.version>2.4.17</akka.version>
  </properties>
    <dependencies>
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
      </dependency>

      <dependency>
          <groupId>com.typesafe.akka</groupId>
          <artifactId>akka-actor_2.11</artifactId>
          <version>${akka.version}</version>
      </dependency>

      <!-- https://mvnrepository.com/artifact/org.scala-lang/scala-actors -->
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-actors</artifactId>
          <version>${scala.actors.version}</version>
      </dependency>

      <dependency>
          <groupId>com.typesafe.akka</groupId>
          <artifactId>akka-remote_2.11</artifactId>
          <version>${akka.version}</version>
      </dependency>

      <!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-client -->
      <dependency>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-client</artifactId>
          <version>2.6.5</version>
      </dependency>
  </dependencies>

用于存放，相应的ActorSystem，Actor的name
//用于存放，相应的ActorSystem，Actor的name

object Constant {
  val RMRS="MyResourceManagerActorSystem"
  val RMA="MyResourceManagerActor"
  val NMAS="MyNodeManagerActorSystem"
  val NMA="MyNodeManagerActor"
}

用于存放相应的receiver中的模式匹配

//注册消息   nodemanager  -> resourcemanager
case class RegisterNodeManager(val nodemanagerid:String,val memory:Int,val cpu:Int)
//注册完成消息 resourcemanager -》 nodemanager
case class RegisteredNodeManager(val resourcemanagerhostname:String)
//用于匹配心跳
case class Heartbeat(val nodemanagerid:String)
//用于存放NodeManagerInfo的相关信息
class NodeManagerInfo(val nodemanagerid:String,val memory:Int,val cpu:Int){
  var lastHeartBeatTime:Long=_
}

//用于匹配超时
case object CheckTimeOut
//用于匹配心跳
case object SendMessage

MyResourceManager

class MyResourceManager(var hostname:String,var port:Int) extends Actor{
  // 用来存储每个注册的NodeManager节点的信息
  private var id2nodemanagerinfo=new mutable.HashMap[String,NodeManagerInfo]()
  // 对所有注册的NodeManager进行去重，其实就是一个HashSet
  private var nodemanagerInfoes = new mutable.HashSet[NodeManagerInfo]()

  // 调度一个任务， 每隔五秒钟执行一次,用于检查是否超时
  override def preStart(): Unit = {
    import scala.concurrent.duration._
    import context.dispatcher
    //也就是发送给receive的相关匹配
    context.system.scheduler.schedule(0 millis,5000 millis,self,CheckTimeOut)
  }

  override def receive: Receive = {
    //用于处理从NodeManager中传来的注册信息
    case RegisterNodeManager(nodemanagerid, memory, cpu)=>{
      val nodeManagerInfo = new NodeManagerInfo(nodemanagerid, memory, cpu)
      // 对注册的NodeManager节点进行存储管理
      id2nodemanagerinfo.put(nodemanagerid,nodeManagerInfo)
      nodemanagerInfoes+=nodeManagerInfo

      //注册完成，将信息，返回到NodeManager
      sender() ! RegisteredNodeManager(hostname+":"+port)
    }
      //注册心跳
    case Heartbeat(nodemanagerid) =>{
      val currentTime = System.currentTimeMillis()
      //获取旧的nodeManagerInfo
      val nodeManagerInfo = id2nodemanagerinfo(nodemanagerid)
      nodeManagerInfo.lastHeartBeatTime=currentTime
      //完成心跳，更新时间
      id2nodemanagerinfo(nodemanagerid)=nodeManagerInfo
      nodemanagerInfoes += nodeManagerInfo
    }
    // 检查过期失效的 NodeManager
    case CheckTimeOut=>{
      val currentTime = System.currentTimeMillis()
      // 15 秒钟失效 并过滤
      nodemanagerInfoes.filter(nm=>currentTime-nm.lastHeartBeatTime>15000)
        .foreach(deadnm=>{
          nodemanagerInfoes -= deadnm
          id2nodemanagerinfo.remove(deadnm.nodemanagerid)
        })
      println("当前注册成功的节点数"+nodemanagerInfoes.size);
    }
  }
}

//注册，创建相应Actor
object MyResourceManager{
  def main(args: Array[String]): Unit = {

    //localhost 9999
    val RESOURCEMANAGER_HOSTNAME=args(0) //主机
    val RESOURCEMANAGER_PORT=args(1).toInt //端口
    val strConfig=
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname =${RESOURCEMANAGER_HOSTNAME}
        |akka.remote.netty.tcp.port=${RESOURCEMANAGER_PORT}
      """.stripMargin
    val conf=ConfigFactory.parseString(strConfig)
    val rmrs: ActorSystem = ActorSystem(Constant.RMRS,conf)
    rmrs.actorOf( Props(new MyResourceManager(RESOURCEMANAGER_HOSTNAME,RESOURCEMANAGER_PORT)),Constant.RMA)
  }
}

MyNodeManager

class MyNodeManager(val resourcemanagerhostname:String, val resourcemanagerport:Int, val memory:Int, val cpu:Int) extends Actor{
  var nodemanagerid:String=_
  var rmRef:ActorSelection=_

  //在启动NodeManager时，向ResourceManager发送注册信息
  override def preStart(): Unit = {
    // 远程path　akka.tcp://（ActorSystem的名称）@（远程地址的IP）:（远程地址的端口）/user/（Actor的名称）
    rmRef=context.actorSelection(s"akka.tcp://${Constant.RMRS}@${resourcemanagerhostname}:${resourcemanagerport}/user/${Constant.RMA}")
    //生产 唯一的nodemanagerid
    nodemanagerid = UUID.randomUUID().toString
    rmRef ! RegisterNodeManager(nodemanagerid,memory,cpu)
  }

  override def receive: Receive = {
    //resourceManager，接受到NodeManager的注册信息后，成功返回给NodeManager信息
    case RegisteredNodeManager(masterURL)=>{
      println(masterURL);
      //发送心跳
      /**
        * initialDelay: FiniteDuration, 多久以后开始执行
        * interval:     FiniteDuration, 每隔多长时间执行一次
        * receiver:     ActorRef, 给谁发送这个消息
        * message:      Any  发送的消息是啥
        */
      import scala.concurrent.duration._
      import context.dispatcher
      context.system.scheduler.schedule(0 millis,4000 millis,self,SendMessage)
    }
    case SendMessage=>{
      //向主节点发送心跳信息
      rmRef ! Heartbeat(nodemanagerid)
      println(Thread.currentThread().getId)
    }

  }
}
object MyNodeManager{
  def main(args: Array[String]): Unit = {
    //localhost localhost 9999 1000 5 8888
    val HOSTNAME=args(0)
    val RM_HOSTNAME =args(1)
    val RM_PORT=args(2).toInt
    val NODEMANAGER_MEMORY = args(3).toInt
    val NODEMANAGER_CORE=args(4).toInt
    var NODEMANAGER_PORT=args(5).toInt
    val strConfig=
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname =${HOSTNAME}
        |akka.remote.netty.tcp.port=${NODEMANAGER_PORT}
      """.stripMargin
    val conf=ConfigFactory.parseString(strConfig)
    val nmas: ActorSystem = ActorSystem(Constant.NMAS,conf)
    nmas.actorOf(Props(new MyNodeManager(RM_HOSTNAME,RM_PORT,NODEMANAGER_MEMORY,NODEMANAGER_CORE)),Constant.NMA)
  }
}

最后启动运行