使用 Akka 实现 Master 与 Worker 之间的通信

MessageProtocol.scala

package top.gldwolf.scala.akkademo.sparkmasterandworker.common/*** @author: Gldwolf* @email: ZengqiangZhao@sina.com* @date: 2020/4/17 10:54*//*** 用于 Work 注册时发送注册信息*/
case class WorkerRegisterInfo(id: String, cpu: Int, ram: Int) {}/*** 用于保存到 Master 的 HashMap 中*/
class WorkerInfo(var id: String, cpu: Int, ram: Int) {var lastHeartBeatTime = System.currentTimeMillis()
}/*** 注册成功后，Master 回应此类型的消息，表示注册成功* Worker 接收到后，启动心跳机制*/
case object RegisteredInfo/*** worker 每隔一定时间由定时器发给自己的一个消息，用于触发自己给 Master 发送消息*/
case object SendHeartBeat/*** 由自己的消息触发，然后给 Master 发送 HeartBeat 消息，消息要带上自己的 id*/
case class HeartBeat(id: String)/*** Master 给自己发送一个触发检查超时 Worker 的消息，定时获取已离线的 Worker*/
case object StartCheckTimeOutWorker/*** Master 给自己发消息，检测 Worker 是否已离线，如果已离线，则移除*/
case object RemoveTimeOutWorker

SparkWorker.scala

package top.gldwolf.scala.akkademo.sparkmasterandworker.workerimport akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import top.gldwolf.scala.akkademo.sparkmasterandworker.common.{HeartBeat, RegisteredInfo, SendHeartBeat, WorkerRegisterInfo}import scala.concurrent.duration.FiniteDuration/*** @author: Gldwolf* @email: ZengqiangZhao@sina.com* @date: 2020/4/17 10:03*/object SparkWorker {def main(args: Array[String]): Unit = {if (args.length < 6) {println("参数个数不正确：host, port, workerName, masterName, masterHost, masterPort...")System.exit(-1)}val host = args(0)val port = args(1).toIntval workerName = args(2)val masterName = args(3)val masterHost = args(4)val masterPort = args(5).toIntval config: Config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port|""".stripMargin)val workerFactory: ActorSystem = ActorSystem("WorkerFactory", config)val workerRef: ActorRef = workerFactory.actorOf(Props(new SparkWorker(masterHost, masterPort, masterName)), workerName)workerRef ! "start"}
}class SparkWorker(masterHost: String, masterPort: Int, masterName: String) extends Actor {var masterRef: ActorSelection = _var id = java.util.UUID.randomUUID().toStringoverride def preStart(): Unit = {masterRef = context.actorSelection("akka.tcp://MasterFactory@" +s"${masterHost}:${masterPort}/user/${masterName}")}override def receive: Receive = {case "start" => {println("Worker " + "已上线~~~")masterRef ! WorkerRegisterInfo(id, 4, 4096)}// 接收到这个消息表示注册成功，然后会给定时给自己发送 SendHeartBeat 消息，触发心跳机制case RegisteredInfo => {println("workerId: " + id + " 注册成功！")import context.dispatcher// 说明：// 1. 0 millis 表示不延时，立即执行定时器// 2. 3000 millis 表示每隔 3 秒执行一次// 3. self 表示发送给自己// 4. SendHeartBeat 表示发送的内容context.system.scheduler.schedule(FiniteDuration(0, "millis"), FiniteDuration(3000, "millis"), self, SendHeartBeat)}// 接收到自己的定时器发送给自己的消息时，触发下面内容，给 Master 发送心跳信息case SendHeartBeat => {println("worker: " + id + " 发送心跳信息~~~")masterRef ! HeartBeat(id)}}
}

SparkMaster.scala

package top.gldwolf.scala.akkademo.sparkmasterandworker.masterimport akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}
import top.gldwolf.scala.akkademo.sparkmasterandworker.common._import scala.collection.mutable
import scala.concurrent.duration.FiniteDuration/*** @author: Gldwolf* @email: ZengqiangZhao@sina.com* @date: 2020/4/17 9:56*/object SparkMaster {def main(args: Array[String]): Unit = {if (args.length < 3) {println("参数不够，请转入 host, port, masterName...")System.exit(-1)}val host = args(0)val port = args(1)val masterName = args(2)val config: Config = ConfigFactory.parseString(s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"|akka.remote.netty.tcp.hostname=$host|akka.remote.netty.tcp.port=$port|""".stripMargin)val masterFactory: ActorSystem = ActorSystem("MasterFactory", config)val masterRef: ActorRef = masterFactory.actorOf(Props[SparkMaster], masterName)masterRef ! "start"}
}class SparkMaster extends Actor {val workers: mutable.HashMap[String, WorkerInfo] = mutable.HashMap()override def receive: Receive = {case "start" => {// 这里开始启动 Master 定时检测任务，判断 Worker 有没有离线println("Master 已上线~~~")self ! StartCheckTimeOutWorker // 给自己发一个开始检测的消息}case WorkerRegisterInfo(id, cpu, ram) => { // 如果是注册信息，则将注册信息管理起来val workerInfo = new WorkerInfo(id, cpu, ram)if (!workers.contains(id)) { // 判断是否已经添加这个 workerworkers += ((id, workerInfo)) // 如果没有则添加进来println("worker：" + id + " 注册成功~~~")println("目前已有的 Workers: " + workers)// 回复注册成功消息sender() ! RegisteredInfo}}case HeartBeat(id) => { // 接收到 worker 的心跳信息val workerInfo = workers(id)val lastHeartBeatTime: Long = System.currentTimeMillis()workerInfo.lastHeartBeatTime = lastHeartBeatTimeprintln("Master 更新了 " + id + " 的心跳时间: " + lastHeartBeatTime)}case StartCheckTimeOutWorker => {// 获取到消息后开始定时检测离线的 Workerprintln("----- Master 开启定时任务，检测已离线的 Worker -----")import context.dispatchercontext.system.scheduler.schedule(FiniteDuration(0, "millis"), FiniteDuration(9000, "millis"), self, RemoveTimeOutWorker)}// 检测哪些 Worker 超时了，并从 Worker 中删除case RemoveTimeOutWorker => {val now: Long = System.currentTimeMillis// 如果最后一次心跳距离现在有 6 秒，那么就代表离线了，则删除这个 Worker
//            for ((id, workerInfo) <- workers) {
//                if ((now - workerInfo.lastHeartBeatTime) / 1000  > 6) {
//                    workers.remove(id)
//                    println("worker: " + id + " 已离线，将其移除！")
//                }
//            }// 也可以使用函数式编程将其移除workers.values.filter(worker => now - worker.lastHeartBeatTime > 6000).foreach(worker => {println("Worker: " + worker.id + " 离线，已将其移除!")workers.remove(worker.id)})println("当前有 " + workers.size + " 个 Worker 存活！")}}
}