服務註冊、發現、心跳

使用 akka 實現，服務註冊、發現、心跳

消息類：

package org.example

case class RegisterWorkerInfo(id: String, cpu: Int, ram: Int);

case class WorkerInfo(val id: String, val cpu: Int, val ram: Int) {
  var lastheartbeat: Long = System.currentTimeMillis();
};

case object RegisteredWorkerInfo;

case object SendHeartBeat;

case object StartTimeOutWorker;

case object RemoveTimeOutWorker;

case class HeartBeanPackage(id: String);

服務端 master：

package org.example

import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration.FiniteDuration

class MasterActor extends Actor {
  val register = mutable.Map[String, WorkerInfo]();

  override def receive: Receive = {
    case "start" => {
      println("master start !!!");
      self ! StartTimeOutWorker;
    }
    case RegisterWorkerInfo(id, cpu, ram) => {
      if (!register.contains(id)) {
        val workerInfo = WorkerInfo(id, cpu, ram);
        register(id) = workerInfo;
      };

      println("workers: " + register)
      sender() ! RegisteredWorkerInfo;
    }
    case HeartBeanPackage(id) => {
      val workerInfo: WorkerInfo = register(id);
      workerInfo.lastheartbeat = System.currentTimeMillis();
      println("master 更新了 " + id + "心跳時間");
    }
    case StartTimeOutWorker =>{
      import context.dispatcher;
      context.system.scheduler.schedule(FiniteDuration(0, TimeUnit.SECONDS), FiniteDuration(9, TimeUnit.SECONDS),
        self, RemoveTimeOutWorker);
    }
    case RemoveTimeOutWorker =>{
      register.filter(x =>  System.currentTimeMillis() -  x._2.lastheartbeat > 6000)
        .foreach(x => register.remove(x._1));
      println("在線 worker數量: " + register.size );
    }
  }
}

object MasterActor {
  def main(args: Array[String]): Unit = {
    val config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname=localhost
         |akka.remote.netty.tcp.port=8080
        """.stripMargin);
    val masterSysterm = ActorSystem("master_systerm", config);
    val masterActor1 = masterSysterm.actorOf(Props[MasterActor], "master_actor1")
    masterActor1 ! "start";
  }
}

客戶端 slaver:

package org.exampleimport java.util.UUIDimport java.util.concurrent.TimeUnitimport akka.actor.{Actor, ActorSelection, ActorSystem, Props}import com.typesafe.config.ConfigFactoryimport scala.concurrent.duration.{FiniteDuration, TimeUnit}class SlaveActor extends Actor {  var masterProxy: ActorSelection = _;  val id: String = UUID.randomUUID().toString;  override def preStart(): Unit = {    val masterUrl = "akka.tcp://master_systerm@localhost:8080/user/master_actor1";    this.masterProxy = context.actorSelection(masterUrl);  }  override def receive: Receive = {    case "start" => {      println("客戶端運行了");      this.masterProxy ! RegisterWorkerInfo(id, 16, 16 * 1024);    }    case RegisteredWorkerInfo => {      println("workerId: " + id + " 註冊成功");      import context.dispatcher;      context.system.scheduler.schedule(FiniteDuration(0, TimeUnit.SECONDS), FiniteDuration(3, TimeUnit.SECONDS), self, SendHeartBeat);    }    case SendHeartBeat =>{      println("workerId: " + id + " 給服務器發送心跳");      masterProxy ! HeartBeanPackage(id);    }  }}object SlaveActor {  def main(args: Array[String]): Unit = {    val config = ConfigFactory.parseString(      s"""         |akka.actor.provider="akka.remote.RemoteActorRefProvider"         |akka.remote.netty.tcp.hostname=localhost         |akka.remote.netty.tcp.port=8090        """.stripMargin);    val slaverSysterm = ActorSystem("slaver_systerm", config);    val slaverActor1 = slaverSysterm.actorOf(Props[SlaveActor], "slaver_actor1")    slaverActor1 ! "start";  }}