nameNode启动过程源码分析

NameNode namenode = createNameNode(argv, null);
if (namenode != null) {
  namenode.join();
}

  FORMAT  ("-format"),
  CLUSTERID ("-clusterid"),
  GENCLUSTERID ("-genclusterid"),
  REGULAR ("-regular"),
  BACKUP  ("-backup"),
  CHECKPOINT("-checkpoint"),
  UPGRADE ("-upgrade"),
  ROLLBACK("-rollback"),
  ROLLINGUPGRADE("-rollingUpgrade"),
  IMPORT  ("-importCheckpoint"),
  BOOTSTRAPSTANDBY("-bootstrapStandby"),
  INITIALIZESHAREDEDITS("-initializeSharedEdits"),
  RECOVER  ("-recover"),
  FORCE("-force"),
  NONINTERACTIVE("-nonInteractive"),
  SKIPSHAREDEDITSCHECK("-skipSharedEditsCheck"),
  RENAMERESERVED("-renameReserved"),
  METADATAVERSION("-metadataVersion"),
  UPGRADEONLY("-upgradeOnly"),
  HOTSWAP("-hotswap"),
  OBSERVER("-observer");

this.haEnabled = HAUtil.isHAEnabled(conf, nsId);
// 检查HA的状态，主要是判断当前启动的是主实例还是备实例
state = createHAState(getStartupOption(conf));
this.allowStaleStandbyReads = HAUtil.shouldAllowStandbyReads(conf);
this.haContext = createHAContext();
try {
  initializeGenericKeys(conf, nsId, namenodeId);
  // 启动NameNode
  initialize(getConf());
  state.prepareToEnterState(haContext);
  try {
    haContext.writeLock();
    state.enterState(haContext);
  } finally {
    haContext.writeUnlock();
  }
} catch (IOException e) {
  this.stopAtException(e);
  throw e;
} catch (HadoopIllegalArgumentException e) {
  this.stopAtException(e);
  throw e;
}

protected void initialize(Configuration conf) throws IOException {
  // .... 省略
  //登录kerberos
  UserGroupInformation.setConfiguration(conf);
  loginAsNameNodeUser(conf);

  // 初始化监控信息
  NameNode.initMetrics(conf, this.getRole());
  StartupProgressMetrics.register(startupProgress);

  pauseMonitor = new JvmPauseMonitor();
  pauseMonitor.init(conf);
  pauseMonitor.start();
  metrics.getJvmMetrics().setPauseMonitor(pauseMonitor);

  // .... 省略

  if (NamenodeRole.NAMENODE == role) {
    startHttpServer(conf);
  }
  // 从本地加载FSImage，并且与Editlog合并产生新的FSImage
  loadNamesystem(conf);
  //TODO 待确认用途
  startAliasMapServerIfNecessary(conf);

  //创建rpcserver，封装了NameNodeRpcServer、ClientRPCServer
  //支持ClientNameNodeProtocol、DataNodeProtocolPB等协议
  rpcServer = createRpcServer(conf);

  initReconfigurableBackoffKey();

  // .... 省略

  if (NamenodeRole.NAMENODE == role) {
    httpServer.setNameNodeAddress(getNameNodeAddress());
    httpServer.setFSImage(getFSImage());
    if (levelDBAliasMapServer != null) {
      httpServer.setAliasMap(levelDBAliasMapServer.getAliasMap());
    }
  }

  //启动执行多个重要的工作线程
  startCommonServices(conf);
  startMetricsLogger(conf);
}

private void startCommonServices(Configuration conf) throws IOException {
  // 创建NameNodeResourceChecker、激活BlockManager等
  namesystem.startCommonServices(conf, haContext);
  registerNNSMXBean();
  if (NamenodeRole.NAMENODE != role) {
    startHttpServer(conf);
    httpServer.setNameNodeAddress(getNameNodeAddress());
    httpServer.setFSImage(getFSImage());
    if (levelDBAliasMapServer != null) {
      httpServer.setAliasMap(levelDBAliasMapServer.getAliasMap());
    }
  }
  // 启动rpc服务
  rpcServer.start();
  try {
    // 获取启动插件列表
    plugins = conf.getInstances(DFS_NAMENODE_PLUGINS_KEY,
        ServicePlugin.class);
  } catch (RuntimeException e) {
    String pluginsValue = conf.get(DFS_NAMENODE_PLUGINS_KEY);
    LOG.error("Unable to load NameNode plugins. Specified list of plugins: " +
        pluginsValue, e);
    throw e;
  }
  // 启动所有插件
  for (ServicePlugin p: plugins) {
    try {
      // 调用插件的start接口，需要插件自己实现，需要实现接口ServicePlugin
      p.start(this);
    } catch (Throwable t) {
      LOG.warn("ServicePlugin " + p + " could not be started", t);
    }
  }
  LOG.info(getRole() + " RPC up at: " + getNameNodeAddress());
  if (rpcServer.getServiceRpcAddress() != null) {
    LOG.info(getRole() + " service RPC up at: "
        + rpcServer.getServiceRpcAddress());
  }
}

void startCommonServices(Configuration conf, HAContext haContext) throws IOException {
  this.registerMBean(); // register the MBean for the FSNamesystemState
  writeLock();
  this.haContext = haContext;
  try {
    //创建NameNodeResourceChecker，并立即检查一次
    nnResourceChecker = new NameNodeResourceChecker(conf);
    checkAvailableResources();
    assert !blockManager.isPopulatingReplQueues();
    StartupProgress prog = NameNode.getStartupProgress();
    prog.beginPhase(Phase.SAFEMODE);
    //获取已完成的数据块总量
    long completeBlocksTotal = getCompleteBlocksTotal();
    prog.setTotal(Phase.SAFEMODE, STEP_AWAITING_REPORTED_BLOCKS,
        completeBlocksTotal);
    // 激活blockManager，blockManager负责管理文件系统中文件的物理块与实际存储位置的映射关系，
    // 是NameNode的核心功能之一。
    blockManager.activate(conf, completeBlocksTotal);
  } finally {
    writeUnlock("startCommonServices");
  }
  
  registerMXBean();
  DefaultMetricsSystem.instance().register(this);
  if (inodeAttributeProvider != null) {
    inodeAttributeProvider.start();
    dir.setINodeAttributeProvider(inodeAttributeProvider);
  }
  // 注册快照管理器
  snapshotManager.registerMXBean();
  InetSocketAddress serviceAddress = NameNode.getServiceAddress(conf, true);
  this.nameNodeHostName = (serviceAddress != null) ? serviceAddress.getHostName() : "";
}

public void activate(Configuration conf, long blockTotal) {
  pendingReconstruction.start();
  // 初始化datanodeManager
  datanodeManager.activate(conf);
  this.redundancyThread.setName("RedundancyMonitor");
  this.redundancyThread.start();
  this.markedDeleteBlockScrubberThread.setName("MarkedDeleteBlockScrubberThread");
  this.markedDeleteBlockScrubberThread.start();
  this.blockReportThread.start();
  mxBeanName = MBeans.register("NameNode", "BlockStats", this);
  bmSafeMode.activate(blockTotal);
}