Hadoop源码分析之NameNode的启动与停止

lshboo

　　在前面几篇文章中已经分析完了IPC的调用的大致过程，接下来分析NameNode节点的工作过程
　　NameNode节点在HDFS文件系统中只有一个实例，它提供了与多个DataNode实例一起组成了HDFS系统。NameNode节点主要功能是维护着HDFS文件中两个重要的关系：

• HDFS文件系统的文件目录树，文件的数据块索引（即每个文件对应的数据块列表），
  
• 数据块和DataNode节点的对应关系（即一个数据块的多个副本分别保存在哪些DataNode节点上）。
  

　　其中HDFS目录树、元信息和数据块索引等信息会持久化到磁盘上，保存在命名空间镜像和编辑日志中。数据块和数据节点的对应关系则在NameNode节点和DataNode节点启动后，由DataNode节点上报给NameNode节点。NameNode节点通过接收数据节点的注册、心跳和数据块提交等信息来管理DataNode节点。NameNode节点会通过发送数据块复制、删除、恢复等NameNode节点指令来操作数据块，此外，NameNode节点还为客户端对文件系统目录树的操作和对文件数据读写以及对HDFS系统进行管理提供了支持。



NameNode的启动
　　在org.apache.hadoop.hdfs.server.namenode.NameNode.java文件中有个main函数，也就是启动NameNode节点的入口函数。NameNode类的main函数中通过方法createNameNode()创建了一个NameNode对象，然后就调用NameNode.join()函数等待NameNode停止。main()函数代码如下：
　　

public static void main(String argv[]) throws Exception {
try {
StringUtils.startupShutdownMessage(NameNode.class, argv, LOG);
NameNode namenode = createNameNode(argv, null);
if (namenode != null)
namenode.join();
} catch (Throwable e) {
LOG.error(StringUtils.stringifyException(e));
System.exit(-1);
}
}NameNode.join()函数就是调用NameNode.server成员变量的join方法，而NameNode.server成员变量是一个Hadoop IPC中讨论的RPC.Server对象，最终就调用了RPC.Server.join()方法。RPC.Server.join()方法也很简单，就是判断Server.runnning变量是否为true，如果running为true，说明这个IPC服务还在运行，就进入等待(wait()函数)。通过上面的分析可以知道，NameNode对象创建成功之后就可以为其他对象提供服务了，直到调用了NameNode的stop()方法，这部分代码比较简单，就不贴出来了。还有一点要注意的是NameNode的成员变量中有两个RPC.Server对象，一个是server，另一个是serviceRpcServer，其中server用于处理客户端的调用，serviceRpcServer用于处理HDFS内部的通信服务，如Datanode节点的调用。　　
　　NameNode类的main()方法中，最重要的就是NameNode namenode = createNameNode(argv, null);这行代码，通过这行代码创建了一个NameNode对象。下面以createNameNode()方法为入口来分析NameNode节点的启动过程。createNameNode()函数的代码如下：
　　

  public static NameNode createNameNode(String argv[],
Configuration conf) throws IOException {
if (conf == null)
conf = new Configuration();
StartupOption startOpt = parseArguments(argv);
if (startOpt == null) {
printUsage();
System.exit(-2);
}
setStartupOption(conf, startOpt);
switch (startOpt) {
case FORMAT:
boolean aborted = format(conf, startOpt.getConfirmationNeeded(),
startOpt.getInteractive());
System.exit(aborted ? 1 : 0);
case FINALIZE:
aborted = finalize(conf, true);
System.exit(aborted ? 1 : 0);
case RECOVER:
NameNode.doRecovery(startOpt, conf);
return null;
default:
}
DefaultMetricsSystem.initialize("NameNode");
NameNode namenode = new NameNode(conf);
return namenode;
}
　　
　　在createNameNode()函数中首先获取一个Configuration对象，用于加载启动NameNode节点所需的配置参数。然后就是调用parseArguments()函数来获取运行参数。如果运行参数有误，那么startOpt变量就为null，调用函数printUsage()打印出NameNode可接受的参数，再结束程序，printUsage()函数打印出的内容如下：




在第一次配置Hadoop环境的时候，都执行过hadoop namenode -fromat命令，这里就是使用到了上面的[-format]参数。　　parseArguments()函数的代码如下：
　　

  private static StartupOption parseArguments(String args[]) {
int argsLen = (args == null) ? 0 : args.length;
StartupOption startOpt = StartupOption.REGULAR;
for(int i=0; i < argsLen; i++) {
String cmd = args;
if (StartupOption.FORMAT.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.FORMAT;
// check if there are other options
for (i = i + 1; i < argsLen; i++) {
if (args.equalsIgnoreCase(StartupOption.FORCE.getName())) {
startOpt.setConfirmationNeeded(false);
}
if (args.equalsIgnoreCase(StartupOption.NONINTERACTIVE.getName())) {
startOpt.setInteractive(false);
}
}
} else if (StartupOption.REGULAR.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.REGULAR;
} else if (StartupOption.UPGRADE.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.UPGRADE;
} else if (StartupOption.RECOVER.getName().equalsIgnoreCase(cmd)) {
if (startOpt != StartupOption.REGULAR) {
throw new RuntimeException(&quot;Can't combine -recover with &quot; +
&quot;other startup options.&quot;);
}
startOpt = StartupOption.RECOVER;
while (++i < argsLen) {
if (args.equalsIgnoreCase(
StartupOption.FORCE.getName())) {
startOpt.setForce(MetaRecoveryContext.FORCE_FIRST_CHOICE);
} else {
throw new RuntimeException(&quot;Error parsing recovery options: &quot; +
&quot;can't understand option \&quot;&quot; + args + &quot;\&quot;&quot;);
}
}
} else if (StartupOption.ROLLBACK.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.ROLLBACK;
} else if (StartupOption.FINALIZE.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.FINALIZE;
} else if (StartupOption.IMPORT.getName().equalsIgnoreCase(cmd)) {
startOpt = StartupOption.IMPORT;
} else
return null;
}
return startOpt;
}首先默认是将startOpt设置位StartupOption.REGULAR，StartupOption是一个枚举类型，由上面打印的Usage可以看出NameNode启动选项主要有5类，分别是format，upgrade，rollback，finalize和importCheckpoint，它的意义分别是：

• format：&#26684;式化NameNode，建立NameNode节点的文件结构。带有format参数启动NameNode节点时，首先启动NameNode节点，然后对其机型&#26684;式化，再关闭节点，如果文件目录已经存在当前文件系统，则会提示用户。它有两个参数nonInteractive和force，nonInteractive表示如果NameNode节点的文件夹在当前的底层文件系统中存在，那么用户将不会收到提示，并且当前的&#26684;式化会失败，force表示不管NameNode的目录存不存在，强制&#26684;式化NameNode节点，也不会提示用户，如果nonInteractive和force参数同时存在，那么force参数将会被忽略；
  
• upgrade：升级系统；
  
• rollback：从升级系统中回滚到前一个版本，这个参数必须在停止的集群分布式文件系统中使用；
  
• finalize：提交一次升级，使用这个参数将会删除前一个版本的文件系统，并且当前版本的文件系统将会变成所使用的文件系统，并且rollback参数将不会再有效，这个参数最终会停止NameNode。
  
• importCheckpoint：从名字节点的一个检查点恢复，这个参数将会从检查点目录导入镜像文件，作为NameNode的文件目录结构。
  
• recover：执行元数据恢复，对于这一点还不理解，HDFS已经可以使用importCheckpoint参数从SecondaryNameNode恢复元数据，为什么还需要一个recover的过程？http://stackoverflow.com/questions/17387477/namenode-recovery-how-does-namenode-recovery-works?answertab=active#tab-top这里说的Hadoop2.0开始可以有多个NameNode节点，而这里是Hadoop1.2.1的源码，这一点留待以后再回过头来分析。
  

　　在startupOption类中，还有一个枚举&#20540;是REGULAR，它表示正常启动NameNode。StartupOption.REGULAR是parseArguments()方法的默认&#20540;，从方法中可以看出，方法会遍历程序运行时给出的所有参数，根据参数&#20540;设定startOpt的&#20540;，如果参数中第一个参数没有NameNode启动所期望的&#20540;，那么就返回null，然后就会打印出启动NameNode的方法。
　　通过parseArguments()方法获取到启动参数之后，就调用私有静态方法将启动方式设置到Congifuration对象中，随后就根据启动参数的&#20540;执行相应的方法，这里&#20540;针对format，finalize和recover的处理。先不管这些处理，看看NameNode对象的创建过程。DefaultMetricsSystem.initialize(&quot;NameNode&quot;);这行代码是用于NameNode的运行时的数据统计的，暂时不考虑。下面就是执行NameNode构造方法创建一个NameNode对象。
　　NameNode的构造方法很简单，直接调用了方法initialize()方法，下面来分析initialize()方法。在这个方法中执行了一系列的成员变量初始化，主要的代码如下(删除了安全相关和度量相关的代码)：
　　

  private void initializea(Configuration conf) throws IOException {
InetSocketAddress socAddr = NameNode.getAddress(conf);
//Handler线程的个数
int handlerCount = conf.getInt(&quot;dfs.namenode.handler.count&quot;, 10);
this.namesystem = new FSNamesystem(this, conf);
// create rpc server，创建IPC服务器
InetSocketAddress dnSocketAddr = getServiceRpcServerAddress(conf);
if (dnSocketAddr != null) {
int serviceHandlerCount =
conf.getInt(DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_KEY,
DFSConfigKeys.DFS_NAMENODE_SERVICE_HANDLER_COUNT_DEFAULT);
this.serviceRpcServer = RPC.getServer(this, dnSocketAddr.getHostName(),
dnSocketAddr.getPort(), serviceHandlerCount,
false, conf, namesystem.getDelegationTokenSecretManager());
this.serviceRPCAddress = this.serviceRpcServer.getListenerAddress();
setRpcServiceServerAddress(conf);
}
this.server = RPC.getServer(this, socAddr.getHostName(),
socAddr.getPort(), handlerCount, false, conf, namesystem
.getDelegationTokenSecretManager());
// Set terse exception whose stack trace won't be logged
this.server.addTerseExceptions(SafeModeException.class);
// The rpc-server port can be ephemeral... ensure we have the correct info
this.serverAddress = this.server.getListenerAddress();
FileSystem.setDefaultUri(conf, getUri(serverAddress));
LOG.info(&quot;Namenode up at: &quot; + this.serverAddress);
//启动http服务器
startHttpServer(conf);
this.server.start();  //start RPC server   
if (serviceRpcServer != null) {
serviceRpcServer.start();      
}
startTrashEmptier(conf);//启动垃圾桶功能
/*可以通过HDFS提供的Service plugin来完成。Service plugin是HDFS提供给管理员的access point，
* 我们可以通过这个access point为HDFS提供一些扩展功能。同样地，创建一个关闭EditLog的plugin。 */
plugins = conf.getInstances(&quot;dfs.namenode.plugins&quot;, ServicePlugin.class);
for (ServicePlugin p: plugins) {
try {
p.start(this);
} catch (Throwable t) {
LOG.warn(&quot;ServicePlugin &quot; + p + &quot; could not be started&quot;, t);
}
}
}首先获取RPC地址，读取xml文件中的属性dfs.namenode.rpc-address的&#20540;，如果所读到的&#20540;为空（不配置默认就为空），则使用fs.default.name属性的&#20540;替代，这个过程相关的代码很清晰。再读取属性dfs.namenode.handler.count的&#20540;，将其赋&#20540;给handlerCount变量，用于配置IPC服务中Handler线程的数量。然后创建FSNamesystem对象。接下来就创建并启动IPC服务器，在NameNode中有两个IPC服务器对象，分别是serviceRpcServer和server。其中如果配置了参数dfs.namenode.servicerpc-address，则会创建serviceRpcServer对象，如果没有配置dfs.namenode.servicerpc-address参数，但是需要创建serviceRpcServer对象，那么就使用dfs.namenode.rpc-address参数创建需要的地址，否则就不创建。serviceRpcServer和server的区别在于，当serviceRpcServer存在时，HDFS内部的IPC调用就使用serviceRpcServer对象来执行，客户端的IPC调用由server对象来执行，如果serviceRpcServer不创建，则所有的IPC调用都由server执行。因为客户端的IPC调用执行时间短，往往需要等待日志的持久化；而数据节点和第二名字节点调用的远程过程，执行逻辑比较复杂，执行时间也比较长。在一个繁忙的系统中，对客户端和HDFS的其他实体的IPC请求使用不同的IPC服务器，可以保证客户端的请求得到即使的响应。　　
　　接下来就调用startHttpServer()方法启动HTTP服务器，调用startTrashEmptier()方法启动垃圾桶功能，最后加载为NameNode配置的插件，由于编译HDFS代码之后就不能修改HDFS代码了，如果要给HDFS增加一些其他功能，可以通过HDFS提供的Service plugin来完成。Service plugin是HDFS提供给管理员的access point，所以可以通过这个access point为HDFS提供一些扩展功能（参考自http://langyu.iyunv.com/blog/1165292）。

Reference
　　《Hadoop技术内幕：深入理解Hadoop Common和HDFS架构设计与实现原理》
　　https://hadoop.apache.org/docs/r1.2.1/commands_manual.html
　　http://langyu.iyunv.com/blog/1165292


　　

版权声明：本文为博主原创文章，未经博主允许不得转载。