Spark2.x精通：Worker节点启动源码剖析

//启动脚本命令
/home/cuadmin/ljs/spark2.2.1/sbin/start-worker.sh
  /日志中打出来的，最后拼写后的命令
 usr/jdk64/jdk1.8.0_112/bin/java -cp home/cuadmin/ljs/spark2.2.1/conf/:/home/cuadmin/ljs/spark2.2.1/jars/* -Xmx1g org.apache.spark.deploy.worker.Worker --webui-port 8081 spark://salver158.hadoop.unicom:7077

def main(argStrings: Array[String]) {
    //同样是初始化了日志对象
    Utils.initDaemon(log)
    val conf = new SparkConf
    //根据用户指定的配置，覆盖默认的配置，如端口、cpu、内存等信息，实例化为args
    val args = new WorkerArguments(argStrings, conf)
    //跟Master节点一样调用startRpcEnvAndEndpoint，创建一个RpcEnv和Endpoint
    val rpcEnv = startRpcEnvAndEndpoint(args.host, args.port, args.webUiPort, args.cores,
      args.memory, args.masters, args.workDir, conf = conf)
    rpcEnv.awaitTermination()
  }

def startRpcEnvAndEndpoint(
      host: String,
      port: Int,
      webUiPort: Int,
      cores: Int,
      memory: Int,
      masterUrls: Array[String],
      workDir: String,
      workerNumber: Option[Int] = None,
      conf: SparkConf = new SparkConf): RpcEnv = {

    // The LocalSparkCluster runs multiple local sparkWorkerX RPC Environments
    val systemName = SYSTEM_NAME + workerNumber.map(_.toString).getOrElse("")
    val securityMgr = new SecurityManager(conf)
    // 创建rpcEnv和endpoint,注意这里创建endpoint用到了Master的ip和端口信息
    // 以及Worker自身分配的cpu、内存信息
    val rpcEnv = RpcEnv.create(systemName, host, port, conf, securityMgr)
    val masterAddresses = masterUrls.map(RpcAddress.fromSparkURL(_))
    rpcEnv.setupEndpoint(ENDPOINT_NAME, new Worker(rpcEnv, webUiPort, cores, memory,
      masterAddresses, ENDPOINT_NAME, workDir, conf, securityMgr))
    rpcEnv
  }

override def onStart() {
    assert(!registered)
    logInfo("Starting Spark worker %s:%d with %d cores, %s RAM".format(
      host, port, cores, Utils.megabytesToString(memory)))
    logInfo(s"Running Spark version ${org.apache.spark.SPARK_VERSION}")
    logInfo("Spark home: " + sparkHome)
   
    //创建Worker工作目录，是伸缩空间和日志输入的目录路径
    //这里有三个相关的参数有兴趣可以了解下，控制目录文件大小：
    //spark.executor.logs.rolling.strategy 
    //spark.executor.logs.rolling.maxSize 134217728 #default byte 
    //spark.executor.logs.rolling.maxRetainedFiles
    createWorkDir()
    //如果用户开启Shffle服务，则启动该服务,   External shuffle Service是长期存在于NodeManager进程中的一个辅助服务。
    // 通过该服务来抓取shuffle数据，减少了Executor的压力，
    // 在Executor GC的时候也不会影响其他Executor的任务运行。
    //这个服务由参数spark.shuffle.service.enabled控制  默认为false 不开启
    shuffleService.startIfEnabled()
   //实例化Worker的UI界面，绑定端口
    webUi = new WorkerWebUI(this, workDir, webUiPort)
    webUi.bind()
    
    workerWebUiUrl = s"http://$publicAddress:${webUi.boundPort}"
    //这里比较重要，向Master进行注册
    registerWithMaster()
    //监控相关，这里不讲解
    metricsSystem.registerSource(workerSource)
    metricsSystem.start()
    // Attach the worker metrics servlet handler to the web ui after the metrics system is started.
    metricsSystem.getServletHandlers.foreach(webUi.attachHandler)
  }

/** Start the external shuffle service */
  def start() {
    require(server == null, "Shuffle server already started")
    val authEnabled = securityManager.isAuthenticationEnabled()
    logInfo(s"Starting shuffle service on port $port (auth enabled = $authEnabled)")
    val bootstraps: Seq[TransportServerBootstrap] =
      if (authEnabled) {
        Seq(new AuthServerBootstrap(transportConf, securityManager))
      } else {
        Nil
      }
    server = transportContext.createServer(port, bootstraps.asJava)

    masterMetricsSystem.registerSource(shuffleServiceSource)
    masterMetricsSystem.start()
  }

private def registerWithMaster() {
    // onDisconnected may be triggered multiple times, so don't attempt registration
    // if there are outstanding registration attempts scheduled.
    registrationRetryTimer match {
      case None =>
        registered = false
        //主要是这行根据master地址和端口，向Master进行注册
        registerMasterFutures = tryRegisterAllMasters()
        connectionAttemptCount = 0
        // 开启一个定时器、如果上面的tryRegisterAllMasters注册失败，那么registered字段就不为TRUE
        //这里就判断registered字段，重试直到最大次数后，放弃重试.
        registrationRetryTimer = Some(forwordMessageScheduler.scheduleAtFixedRate(
          new Runnable {
            override def run(): Unit = Utils.tryLogNonFatalError {
              Option(self).foreach(_.send(ReregisterWithMaster))
            }
          },
          INITIAL_REGISTRATION_RETRY_INTERVAL_SECONDS,
          INITIAL_REGISTRATION_RETRY_INTERVAL_SECONDS,
          TimeUnit.SECONDS))
      case Some(_) =>
        logInfo("Not spawning another attempt to register with the master, since there is an" +
          " attempt scheduled already.")
    }
  }

private def tryRegisterAllMasters(): Array[JFuture[_]] = {
    //循环该worker的构造参数属性masterRpcAddresses，这是一个存放着RpcAddress的集合
    //然后使用一个线程池（个数=masterRpcAddresses个数）来多线程注册Master
    //这里向所有Master进行注册，由Master的代码我们知道，只有Active的Master才会返回消息，其他不返回消息
    //这个masterRpcAddresses是在startRpcEnvAndEndpoint函数中调用rpcEnv.setupEndpoint的时候
    //会new Worker传进来的参数
    masterRpcAddresses.map { masterAddress =>
      registerMasterThreadPool.submit(new Runnable {
        override def run(): Unit = {
          try {
            logInfo("Connecting to master " + masterAddress + "...")
            //根据masterEndPoint地址 和名称 向Master进行注册
            val masterEndpoint = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
            sendRegisterMessageToMaster(masterEndpoint)
          } catch {
            case ie: InterruptedException => // Cancelled
            case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
          }
        }
      })
    }
  }

 private def sendRegisterMessageToMaster(masterEndpoint: RpcEndpointRef): Unit = {
   // 发送消息 这里参数有唯一的id， 主机名 端口 cpu 内存等信息
    masterEndpoint.send(RegisterWorker(
      workerId,
      host,
      port,
      self,
      cores,
      memory,
      workerWebUiUrl,
      masterEndpoint.address))
  }

 changeMaster(masterRef, masterWebUiUrl, masterAddress)
        forwordMessageScheduler.scheduleAtFixedRate(new Runnable {
          override def run(): Unit = Utils.tryLogNonFatalError {
            self.send(SendHeartbeat)
          }
        }, 0, HEARTBEAT_MILLIS, TimeUnit.MILLISECONDS)

 forwordMessageScheduler.scheduleAtFixedRate(new Runnable {
            override def run(): Unit = Utils.tryLogNonFatalError {
              self.send(WorkDirCleanup)
            }
          }, CLEANUP_INTERVAL_MILLIS, CLEANUP_INTERVAL_MILLIS, TimeUnit.MILLISECONDS)