spark之master与worker通信模型讲解

通信模型架构图

master 端代码
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
// 需要导入这2个包 封装一些属性。
class MasterActor extends Actor {

  //在开始之前调用一次
  override def preStart(): Unit = {

  }

  //用于接收消息
  override def receive: Receive = {
    case "started" => {
      println("Master has been started!")
      //进入这个分支，说明这个Master线程已经启动完成
    }
    case "connecting" => {
      println("Master has been get connect from Worker!")
      println("a Worker Node has been register!")
      //返回消息给Worker
      sender() ! "connected"
      Thread.sleep(1000)
    }

    case "stoped" => {

    }
  }

}

object Demo01MasterActor {

  def main(args: Array[String]) {

    //设置MasterIP和端口
    val masterHost = "localhost"
    val masterPort = "1234"

    //端口和IP封装到akka架构，获取一个属性配置文件
    val conStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$masterHost"
         |akka.remote.netty.tcp.port = "$masterPort"
  """.stripMargin

    val config = ConfigFactory.parseString(conStr)
    val masterActorSystem = ActorSystem("MasterActorSystem", config)
    val masterActor = masterActorSystem.actorOf(Props[MasterActor], "MasterActor")
    masterActor ! "started"
    masterActorSystem.awaitTermination();

  }

}

worker端代码
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

class WorkerActor extends Actor {
  var masterURL: ActorSelection = null

  //启动Actor之前执行，做初始化工作
  override def preStart(): Unit = {
    //配置访问Master的URL
    //MasterIP:localhost
    //MasterPort:8888（根据Master配置）
    //Master的 ActorSystem对象：MasterActorSystem、MasterActor
    masterURL = context.actorSelection("akka.tcp://[email protected]:8888/user/MasterActor")
  }

  override def receive: Receive = {
    case "started" => {
      println("Worker has been started!")
      //进入这个分支，说明这个Worker线程已经启动完成
      //可以去向Master注册

      //请求和Master建立连接
      masterURL ! "connecting"
    }
    case "connected" => {
      println("Worker 收到来自Master确认信息!")
    }
    case "stoped" => {

    }
  }

}

object Demo01WorkerActor {

  def main(args: Array[String]) {
    //初始化MastereIP和端口、WorkerIP和端口

    //    val masterHost = args(0)
    //    val masterPort = args(1)
    //    val workerHost = args(2)
    //    val workePort = args(3)

    val masterHost = "localhost"
    val masterPort = "8888"

    val workerHost = "localhost"
    val workePort = "8889"

    //端口和IP封装到akka架构，获取一个属性配置文件
    val conStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$workerHost"
         |akka.remote.netty.tcp.port = "$workePort"
  """.stripMargin

    val config = ConfigFactory.parseString(conStr)
    val workerActorSystem = ActorSystem("WorkerActorSystem", config)
    val workerActor = workerActorSystem.actorOf(Props[WorkerActor], "WorkerActor")
    workerActor ! "started"
    workerActorSystem.awaitTermination();

  }

}