正版复古传奇手游官网,单职业传奇手游版官网新开

概述

Worker的启动都是通过启动shell脚本

Master启动

master启动从main函数开始,主要启动Rpc环境：RpcEnv（Rpc环境）：Akka和Netty

启动一个Master，通过启动 Shell 脚本start-master.sh

这个脚本实际启动 spark 的 master 类

start-master.sh? -> spark-daemon.sh start org.apache.spark.deploy.master.Master

启动时会传入一些参数，比如cpu的执行核数，内存大小，app的main方法等

查看Master类的main方法

private[spark] object Master extends Logging {

? val systemName = "sparkMaster"

? private val actorName = "Master"

? //master启动的入口，启动命令里会传入一些参数

? def main(argStrings: Array[String]) {

? ? SignalLogger.register(log)

? ? //创建SparkConf? ? val conf = new SparkConf

? ? //保存参数到SparkConf

? ? val args = new MasterArguments(argStrings, conf)

? ? //创建ActorSystem

? ? val (actorSystem, _, _, _) = startSystemAndActor(args.host, args.port, args.webUiPort, conf)

? ? //等待该主Actor结束

? ? actorSystem.awaitTermination()

? }

这里主要看startSystemAndActor方法

? /**

? *? (1) 启动Master的actor system

? *? (2) 绑定端口

? *? (3) 启动webui和port

? *? (4) 启动rest服务和绑定端口

? */

? def startSystemAndActor(

? ? ? host: String,

? ? ? port: Int,

? ? ? webUiPort: Int,

? ? ? conf: SparkConf): (ActorSystem, Int, Int, Option[Int]) = {

? ? val securityMgr = new SecurityManager(conf)

? ? //利用AkkaUtils创建ActorSystem

? ? val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port, conf = conf,

? ? ? securityManager = securityMgr)

? ? val actor = actorSystem.actorOf(

? ? ? Props(classOf[Master], host, boundPort, webUiPort, securityMgr, conf), "Master")

? ....

? }

}

spark底层通信是Akka

通过ActorSystem创建Actor -> actorSystem.actorOf, 就会执行Master的构造方法（也就是说上面调用actorOf方法的时候会创建actor，也就是调用Master的构造器）->然后执行Actor生命周期方法

执行Master的构造方法初始化一些变量

private[spark] class Master(

? ? host: String,

? ? port: Int,

? ? webUiPort: Int,

? ? val securityMgr: SecurityManager,

? ? val conf: SparkConf)

? extends Actor with ActorLogReceive with Logging with LeaderElectable {

? //主构造器

? //启用定期器功能

? import context.dispatcher? // to use Akka's scheduler.schedule()

? val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)

? def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")? // For application IDs

? //woker超时时间

? val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000

? val RETAINED_APPLICATIONS = conf.getInt("spark.deploy.retainedApplications", 200)

? val RETAINED_DRIVERS = conf.getInt("spark.deploy.retainedDrivers", 200)

? val REAPER_ITERATIONS = conf.getInt("spark.dead.worker.persistence", 15)

? val RECOVERY_MODE = conf.get("spark.deploy.recoveryMode", "NONE")

? //一个HashSet用于保存WorkerInfo

? val workers = new HashSet[WorkerInfo]

? //一个HashMap用保存workid -> WorkerInfo

? val idToWorker = new HashMap[String, WorkerInfo]

? val addressToWorker = new HashMap[Address, WorkerInfo]

? //一个HashSet用于保存客户端（SparkSubmit）提交的任务

? val apps = new HashSet[ApplicationInfo]

? //一个HashMap Appid-》 ApplicationInfo

? val idToApp = new HashMap[String, ApplicationInfo]

? val actorToApp = new HashMap[ActorRef, ApplicationInfo]

? val addressToApp = new HashMap[Address, ApplicationInfo]

? //等待调度的App

? val waitingApps = new ArrayBuffer[ApplicationInfo]

? val completedApps = new ArrayBuffer[ApplicationInfo]

? var nextAppNumber = 0

? val appIdToUI = new HashMap[String, SparkUI]

? //保存DriverInfo

? val drivers = new HashSet[DriverInfo]

? val completedDrivers = new ArrayBuffer[DriverInfo]

? val waitingDrivers = new ArrayBuffer[DriverInfo] // Drivers currently spooled for scheduling

主构造器执行完就会执行preStart –》执行完receive方法

? //启动定时器，进行定时检查超时的worker

? //重点看一下CheckForWorkerTimeOut

? context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut)

? ? ? 1、在第一次运行的时候需要等待多少时间；

　　2、循环的频率；

　　3、我们想发送消息的目标ActorRef ；

　　4、消息

preStart方法里创建了一个定时器，定时检查Woker的超时时间val WORKER_TIMEOUT = conf.getLong("spark.worker.timeout", 60) * 1000默认为60秒

到此Master的初始化的主要过程到我们已经看到了，主要就是构造一个Master的Actor进行等待消息，并初始化了集合来保存task信息和Worker信息，和一个定时器来检查Worker的超时

Woker的启动

执行本地 shell 脚本salves.sh-> 通过读取配置文件，通过ssh的方式远程连接远端的worker节点，然后启动每个节点的 work 类

spark-daemon.sh start org.apache.spark.deploy.worker.Worker

脚本会启动org.apache.spark.deploy.worker.Worker 类

看Worker源码：

private[spark] object Worker extends Logging {

? //Worker启动的入口

? def main(argStrings: Array[String]) {

? ? SignalLogger.register(log)

? ? val conf = new SparkConf

? ? val args = new WorkerArguments(argStrings, conf)

? ? //新创ActorSystem和Actor

? ? val (actorSystem, _) = startSystemAndActor(args.host, args.port, args.webUiPort, args.cores,

? ? ? args.memory, args.masters, args.workDir)

? ? actorSystem.awaitTermination()

? }

这里最重要的是Woker的startSystemAndActor

? def startSystemAndActor(

? ? ? host: String,

? ? ? port: Int,

? ? ? webUiPort: Int,

? ? ? cores: Int,

? ? ? memory: Int,

? ? ? masterUrls: Array[String],

? ? ? workDir: String,

? ? ? workerNumber: Option[Int] = None,

? ? ? conf: SparkConf = new SparkConf): (ActorSystem, Int) = {

? ? // The LocalSparkCluster runs multiple local sparkWorkerX actor systems

? ? val systemName = "sparkWorker" + workerNumber.map(_.toString).getOrElse("")

? ? val actorName = "Worker"

? ? val securityMgr = new SecurityManager(conf)

? ? //通过AkkaUtils ActorSystem

? ? val (actorSystem, boundPort) = AkkaUtils.createActorSystem(systemName, host, port,

? ? ? conf = conf, securityManager = securityMgr)

? ? val masterAkkaUrls = masterUrls.map(Master.toAkkaUrl(_, AkkaUtils.protocol(actorSystem)))

? ? //通过actorSystem.actorOf创建Actor? Worker-》执行构造器 -》 preStart -》 receice

? ? actorSystem.actorOf(Props(classOf[Worker], host, boundPort, webUiPort, cores, memory,

? ? ? masterAkkaUrls, systemName, "Worker",? workDir, conf, securityMgr), name = "Worker")

? ? (actorSystem, boundPort)

? }

这里启动该Worker的Actor对象,到此Worker的启动初始化完成

Worker与Master通信

根据Actor生命周期接着Worker的preStart方法被调用，也就是说worker一起动就会给master发消息，进行注册（说白了就是把work信息存到master的一个list里）

? override def preStart() {

? ? assert(!registered)

? ? createWorkDir()

? ? context.system.eventStream.subscribe(self, classOf[RemotingLifecycleEvent])

? ? shuffleService.startIfEnabled()

? ? webUi = new WorkerWebUI(this, workDir, webUiPort)

? ? webUi.bind()

? ? //Worker向Master注册

? ? registerWithMaster()

? ? ....

? }

这里调用了一个registerWithMaster方法，开始向Master注册

def registerWithMaster() {

? ? // DisassociatedEvent may be triggered multiple times, so don't attempt registration

? ? // if there are outstanding registration attempts scheduled.

? ? registrationRetryTimer match {

? ? ? case None =>

? ? ? ? registered = false

? ? ? ? //开始注册

? ? ? ? tryRegisterAllMasters()

? ? ? ? ....

? ? }

? }

registerWithMaster里通过匹配调用了tryRegisterAllMasters方法

，接下来看

? private def tryRegisterAllMasters() {

? ? //遍历master的地址

? ? for (masterAkkaUrl <- masterAkkaUrls) {

? ? ? logInfo("Connecting to master " + masterAkkaUrl + "...")

? ? ? //Worker得到Mater actor的远程引用? ? ? val actor = context.actorSelection(masterAkkaUrl)

? ? ? //向Master发送注册信息

? ? ? actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)//Worker向Master发送了一个消息，注册内容包含,带去一些参数，id,主机，端口，cpu核数，内存等待? ? }

? }

通过masterAkkaUrl和Master建立连接后

masterActor接受来自Worker的注册信息

override def receiveWithLogging = {

? ? ......

? ? //接受来自Worker的注册信息

? ? case RegisterWorker(id, workerHost, workerPort, cores, memory, workerUiPort, publicAddress) =>

? ? {

? ? ? logInfo("Registering worker %s:%d with %d cores, %s RAM".format(

? ? ? ? workerHost, workerPort, cores, Utils.megabytesToString(memory)))

? ? ? if (state == RecoveryState.STANDBY) {

? ? ? ? // ignore, don't send response

? ? ? ? //判断这个worker是否已经注册过

? ? ? } else if (idToWorker.contains(id)) {

? ? ? ? //如果注册过，告诉worker注册失败

? ? ? ? sender ! RegisterWorkerFailed("Duplicate worker ID")

? ? ? } else {

? ? ? ? //没有注册过，把来自Worker的注册信息封装到WorkerInfo当中

? ? ? ? val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,

? ? ? ? ? sender, workerUiPort, publicAddress)

? ? ? ? if (registerWorker(worker)) {

? ? ? ? ? //用持久化引擎记录Worker的信息

? ? ? ? ? persistenceEngine.addWorker(worker)

? ? ? ? ? //向Worker反馈信息，告诉Worker注册成功

? ? ? ? ? sender ! RegisteredWorker(masterUrl, masterWebUiUrl)

? ? ? ? ? schedule()

? ? ? ? } else {

? ? ? ? ? val workerAddress = worker.actor.path.address

? ? ? ? ? logWarning("Worker registration failed. Attempted to re-register worker at same " +

? ? ? ? ? ? "address: " + workerAddress)

? ? ? ? ? sender ! RegisterWorkerFailed("Attempted to re-register worker at same address: "

? ? ? ? ? ? + workerAddress)

? ? ? ? }

? ? ? }

? ? }

注册成功后Worker向master发送心跳

override def receiveWithLogging = {

? ? ? case RegisteredWorker(masterUrl, masterWebUiUrl) =>

? ? ? logInfo("Successfully registered with master " + masterUrl)

? ? ? registered = true

? ? ? changeMaster(masterUrl, masterWebUiUrl)

? ? ? //启动定时器，定时发送心跳Heartbeat

? ? ? context.system.scheduler.schedule(0 millis, HEARTBEAT_MILLIS millis, self, SendHeartbeat)

? ? ? if (CLEANUP_ENABLED) {

? ? ? ? logInfo(s"Worker cleanup enabled; old application directories will be deleted in: $workDir")

? ? ? ? context.system.scheduler.schedule(CLEANUP_INTERVAL_MILLIS millis,

? ? ? ? ? CLEANUP_INTERVAL_MILLIS millis, self, WorkDirCleanup)

? ? ? }

worker接受来自Master的注册成功的反馈信息,启动定时器，定时发送心跳Heartbeat

? ? case SendHeartbeat =>

? ? ? //worker发送心跳的目的就是为了报活

? ? ? if (connected) { master ! Heartbeat(workerId) }

Master接收心跳消息，更新最后一次心跳时间

? override def receiveWithLogging = {

? ? ? ? ....

? ? case Heartbeat(workerId) => {

? ? ? idToWorker.get(workerId) match {

? ? ? ? case Some(workerInfo) =>

? ? ? ? ? //更新最后一次心跳时间

? ? ? ? ? workerInfo.lastHeartbeat = System.currentTimeMillis()

? ? ? ? ? .....

? ? ? }

? ? }

}

记录并更新workerInfo.lastHeartbeat = System.currentTimeMillis()最后一次心跳时间

Master的定时任务会不断的发送一个CheckForWorkerTimeOut内部消息不断的轮询集合里的Worker信息，如果超过60秒就将Worker信息移除

? //检查超时的Worker

? ? case CheckForWorkerTimeOut => {

? ? ? timeOutDeadWorkers()

? ? }

timeOutDeadWorkers方法

? def timeOutDeadWorkers() {

? ? // Copy the workers into an array so we don't modify the hashset while iterating through it

? ? val currentTime = System.currentTimeMillis()

? ? val toRemove = workers.filter(_.lastHeartbeat < currentTime - WORKER_TIMEOUT).toArray

? ? for (worker <- toRemove) {

? ? ? if (worker.state != WorkerState.DEAD) {

? ? ? ? logWarning("Removing %s because we got no heartbeat in %d seconds".format(

? ? ? ? ? worker.id, WORKER_TIMEOUT/1000))

? ? ? ? removeWorker(worker)

? ? ? } else {

? ? ? ? if (worker.lastHeartbeat < currentTime - ((REAPER_ITERATIONS + 1) * WORKER_TIMEOUT)) {

? ? ? ? ? workers -= worker // we've seen this DEAD worker in the UI, etc. for long enough; cull it

? ? ? ? }

? ? ? }

? ? }

? }

如果（最后一次心跳时间<当前时间-超时时间）则判断为Worker超时，

将集合里的信息移除。

当下一次收到心跳信息时，如果是已注册过的，workerId不为空，但是WorkerInfo已被移除的条件，就会sender ! ReconnectWorker(masterUrl)发送一个重新注册的消息

case None =>

? ? ? ? ? if (workers.map(_.id).contains(workerId)) {

? ? ? ? ? ? logWarning(s"Got heartbeat from unregistered worker $workerId." +

? ? ? ? ? ? ? " Asking it to re-register.")

? ? ? ? ? ? //发送重新注册的消息

? ? ? ? ? ? sender ! ReconnectWorker(masterUrl)

? ? ? ? ? } else {

? ? ? ? ? ? logWarning(s"Got heartbeat from unregistered worker $workerId." +

? ? ? ? ? ? ? " This worker was never registered, so ignoring the heartbeat.")

? ? ? ? ? }

Master与Worker启动的大致的通信流程到此ok

传奇手游全部平台_三端传奇开服网址大全下载_三端传奇版本下载教程

spark-源码 master和worker启动

spark-源码 master和worker启动