学步园

Hadoop集群中，不同进程之间通信需要使用合适的协议才能够进行交互，之前对Hadoop给出的协议接口做了分析。在协议接口中约定了通信双方的特定行为，那么，在实现这些通信协议的实现类中，就能看到指定进程是如何实现协议接口中约定的行为的。这里，阅读分析org.apache.hadoop.hdfs.server.namenode.Namenode实现类。

首先，看一下Namenode类实现的接口，下面是该类声明：

public class NameNode implements ClientProtocol, DatanodeProtocol, NamenodeProtocol, FSConstants, RefreshAuthorizationPolicyProtocol;

可以看到，Namenode主要实现了ClientProtocol，DatanodeProtocol，NamenodeProtocol这三个用来通信的协议。其中，RefreshAuthorizationPolicyProtocol接口是定义所使用的认证策略，并能根据不同的应用场景来自动刷新其级别，以适应实际应用的需要。

Namenode是HDFS集群中的中心服务器，对于服务器的配置选项，可以通过加载配置文件来进行配置，所以该类中有如下加载配置资源的两行代码：

static{<br /> Configuration.addDefaultResource("hdfs-default.xml"); // 默认配置文件<br /> Configuration.addDefaultResource("hdfs-site.xml"); // 用户配置文件<br /> }

Namenode类中定义属性如下：

public static final int DEFAULT_PORT = 8020; // 默认端口号<br /> public FSNamesystem namesystem; // 用来维护与Datanode相关的重要列表<br /> private Server server; // RPC服务器<br /> private InetSocketAddress serverAddress = null; // RPC服务器地址<br /> private HttpServer httpServer; // 内嵌的HTTP服务器，用来应答到来的HTTP请求<br /> private InetSocketAddress httpAddress = null; // HTTP服务器地址<br /> private Thread emptier; // 回收站清理线程<br /> private boolean stopRequested = false; // 是否阻止请求<br /> private boolean serviceAuthEnabled = false; // 是否启动服务认证级别</p> <p> static NameNodeMetrics myMetrics; // 用来维护Namenode服务器活动状况的统计数据实体

对上面几个重要的属性简单说明一下。

FSNamesystem也是org.apache.hadoop.hdfs.server.namenode包中的类，在Namenode类中是核心的、最重要的。该类主要的功能是对Datanode结点的一些状态等进行登记，便于Namenode进程能够快速获取到指定的Datanode结点的状态等的详细信息，以便进行任务的调度与分配。它主要跟踪如下几个重要的数据表：

1）文件系统到块的映射表（有效文件系统名字-->块列表）；

2）全部有效块的列表；

3）块到主机的映射表（块-->主机列表）；

4）主机到块的映射表（主机-->块列表）；

5）LRU缓存（已经更新心跳状态的主机都放到LRU Cache中）。

HttpServer在org.apache.hadoop.http包中，提供了将Jetty服务器内置于Namenode服务器中，以便可以通过HTTP请求的方式来获取当前Namenode提供服务的信息。

Namenode提供了一个执行初始化的方法，如下所示：

/**<br /> * 初始化Namenode<br /> */<br /> private void initialize(Configuration conf) throws IOException {<br /> InetSocketAddress socAddr = NameNode.getAddress(conf); // 从配置conf中获取到Namenode服务器所使用的Socket地址<br /> int handlerCount = conf.getInt("dfs.namenode.handler.count", 10); // 服务器上处理器Handler线程的数量</p> <p> if (serviceAuthEnabled = conf.getBoolean(erviceAuthorizationManager.SERVICE_AUTHORIZATION_CONFIG, false)) {<br /> PolicyProvider policyProvider =<br /> (PolicyProvider)(ReflectionUtils.newInstance(conf.getClass(PolicyProvider.POLICY_PROVIDER_CONFIG, HDFSPolicyProvider.class, PolicyProvider.class), conf));<br /> SecurityUtil.setPolicy(new ConfiguredPolicy(conf, policyProvider)); // 设置安全策略<br /> }</p> <p> this.server = RPC.getServer(this, socAddr.getHostName(), socAddr.getPort(), handlerCount, false, conf); // 创建RPC服务器</p> <p> this.serverAddress = this.server.getListenerAddress();<br /> FileSystem.setDefaultUri(conf, getUri(serverAddress));<br /> LOG.info("Namenode up at: " + this.serverAddress);</p> <p> myMetrics = new NameNodeMetrics(conf, this); // 初始化NameNodeMetrics</p> <p> this.namesystem = new FSNamesystem(this, conf);<br /> startHttpServer(conf);<br /> this.server.start(); // 启动RPC服务器<br /> startTrashEmptier(conf);<br /> }

在Namenode类实现中，其中实现的操作基本上都是由FSNamesystem来完成的。在这里，我们先不关心Namenode具体是如何实现那些基本操作的，而只是关注Namenode的功能特性，在后面再对FSNamesystem类进行分析。这里使用方式就是，根据Namenode所实现的接口中定义的操作，来分析Namenode服务器所具备的基本功能，或者说提供的基本服务。

NamenodeProtocol实现

在Namenode类中，实现了NamenodeProtocol协议接口中定义的getBlocks方法，如下所示：

/**<br /> * 返回指定大小为size的Datanode上块及其所在位置列表<br /> */<br /> public BlocksWithLocations getBlocks(DatanodeInfo datanode, long size) throws IOException {<br /> if(size <= 0) {<br /> throw new IllegalArgumentException("Unexpected not positive size: "+size);<br /> }<br /> return namesystem.getBlocks(datanode, size); // 调用namesystem的getBlocks方法来真正实现<br /> }

给定DatanodeInfo datanode，它是一个描述Datanode的状态的实体类对象，通过getBlocks方法，可以获取到总的块大小为size的BlocksWithLocations，即描述这些块的位置信息的BlocksWithLocations对象。

ClientProtocol实现

实现ClientProtocol协议接口中定义的方法，如下所示：

  /**<br /> * 获取到指定文件src的全部块的信息<br /> *<br /> * @param src 指定的文件<br /> * @param offset 起始偏移位置<br /> * @param length 长度<br /> */<br /> public LocatedBlocks getBlockLocations(String src, long offset, long length) throws IOException {<br /> myMetrics.numGetBlockLocations.inc();<br /> return namesystem.getBlockLocations(getClientMachine(), src, offset, length); // 调用：获取到指定文件的全部块信息<br /> }</p> <p> /**<br /> * 在制定的文件系统命名空间中创建一个文件入口（entry）<br /> */<br /> public void create(String src, FsPermission masked, String clientName, boolean overwrite, short replication, long blockSize) throws IOException {<br /> String clientMachine = getClientMachine();<br /> if (stateChangeLog.isDebugEnabled()) {<br /> stateChangeLog.debug("*DIR* NameNode.create: file " +src+" for "+clientName+" at "+clientMachine);<br /> }<br /> if (!checkPathLength(src)) {<br /> throw new IOException("create: Pathname too long. Limit " + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");<br /> }<br /> namesystem.startFile(src,<br /> new PermissionStatus(UserGroupInformation.getCurrentUGI().getUserName(),<br /> null, masked),<br /> clientName, clientMachine, overwrite, replication, blockSize); // 调用：执行文件的创建操作<br /> myMetrics.numFilesCreated.inc();<br /> myMetrics.numCreateFileOps.inc();<br /> }</p> <p> /**<br /> * 对指定文件执行追加写操作<br /> */<br /> public LocatedBlock append(String src, String clientName) throws IOException {<br /> String clientMachine = getClientMachine();<br /> if (stateChangeLog.isDebugEnabled()) {<br /> stateChangeLog.debug("*DIR* NameNode.append: file " +src+" for "+clientName+" at "+clientMachine);<br /> }<br /> LocatedBlock info = namesystem.appendFile(src, clientName, clientMachine); // 调用：执行文件的追加写操作<br /> myMetrics.numFilesAppended.inc();<br /> return info;<br /> }</p> <p> /**<br /> * 设置副本因子<br /> */<br /> public boolean setReplication(String src, short replication) throws IOException {<br /> return namesystem.setReplication(src, replication); // 调用：设置副本因子<br /> }</p> <p> /**<br /> * 设置权限<br /> */<br /> public void setPermission(String src, FsPermission permissions) throws IOException {<br /> namesystem.setPermission(src, permissions); // 调用：设置权限<br /> }</p> <p> /**<br /> * 设置文件属主<br /> */<br /> public void setOwner(String src, String username, String groupname) throws IOException {<br /> namesystem.setOwner(src, username, groupname); // 调用：设置文件属主<br /> }</p> <p> /**<br /> * 向指定的文件中写入数据块<br /> */<br /> public LocatedBlock addBlock(String src, String clientName) throws IOException {<br /> stateChangeLog.debug("*BLOCK* NameNode.addBlock: file " +src+" for "+clientName);<br /> LocatedBlock locatedBlock = namesystem.getAdditionalBlock(src, clientName); // 调用：获取并执行写入操作<br /> if (locatedBlock != null)<br /> myMetrics.numAddBlockOps.inc();<br /> return locatedBlock;<br /> }</p> <p> /**<br /> * 客户端放弃对指定块的操作<br /> */<br /> public void abandonBlock(Block b, String src, String holder) throws IOException {<br /> stateChangeLog.debug("*BLOCK* NameNode.abandonBlock: " +b+" of file "+src);<br /> if (!namesystem.abandonBlock(b, src, holder)) { // 调用<br /> throw new IOException("Cannot abandon block during write to " + src);<br /> }<br /> }</p> <p> /**<br /> * 客户端完成对指定文件的写操作<br /> */<br /> public boolean complete(String src, String clientName) throws IOException {<br /> stateChangeLog.debug("*DIR* NameNode.complete: " + src + " for " + clientName);<br /> CompleteFileStatus returnCode = namesystem.completeFile(src, clientName); // 调用：执行写操作，写完成之后关闭该文件流<br /> if (returnCode == CompleteFileStatus.STILL_WAITING) {<br /> return false;<br /> } else if (returnCode == CompleteFileStatus.COMPLETE_SUCCESS) {<br /> return true;<br /> } else {<br /> throw new IOException("Could not complete write to file " + src + " by " + clientName);<br /> }<br /> }</p> <p> /**<br /> * 客户端向Namenode报告corrupted块的信息<br /> */<br /> public void reportBadBlocks(LocatedBlock[] blocks) throws IOException {<br /> stateChangeLog.info("*DIR* NameNode.reportBadBlocks");<br /> for (int i = 0; i < blocks.length; i++) {<br /> Block blk = blocks[i].getBlock();<br /> DatanodeInfo[] nodes = blocks[i].getLocations();<br /> for (int j = 0; j < nodes.length; j++) {<br /> DatanodeInfo dn = nodes[j];<br /> namesystem.markBlockAsCorrupt(blk, dn); // 调用：标记属于该Datanode的corrupted状态的块<br /> }<br /> }<br /> }</p> <p> /**<br /> * 重命名文件<br /> */<br /> public boolean rename(String src, String dst) throws IOException {<br /> stateChangeLog.debug("*DIR* NameNode.rename: " + src + " to " + dst);<br /> if (!checkPathLength(dst)) {<br /> throw new IOException("rename: Pathname too long. Limit " + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");<br /> }<br /> boolean ret = namesystem.renameTo(src, dst); // 调用：重命名文件<br /> if (ret) {<br /> myMetrics.numFilesRenamed.inc();<br /> }<br /> return ret;<br /> }</p> <p> /**<br /> * 删除文件<br /> */<br /> public boolean delete(String src, boolean recursive) throws IOException {<br /> if (stateChangeLog.isDebugEnabled()) {<br /> stateChangeLog.debug("*DIR* Namenode.delete: src=" + src + ", recursive=" + recursive);<br /> }<br /> boolean ret = namesystem.delete(src, recursive); // 调用：删除文件<br /> if (ret)<br /> myMetrics.numDeleteFileOps.inc();<br /> return ret;<br /> }</p> <p> /**<br /> * 管理客户端状态，如果Namenode在一段时间内接收不到客户端心跳状态，标记为挂掉；经过周期检测，又获取到对应客户端心跳状态，为其解锁。<br /> */<br /> public void renewLease(String clientName) throws IOException {<br /> namesystem.renewLease(clientName); // 调用<br /> }</p> <p> /**<br /> * 获取指定目录下文件列表<br /> */<br /> public FileStatus[] getListing(String src) throws IOException {<br /> FileStatus[] files = namesystem.getListing(src); // 调用<br /> if (files != null) {<br /> myMetrics.numGetListingOps.inc();<br /> }<br /> return files;<br /> }</p> <p> /**<br /> * 获取指定文件的状态信息<br /> */<br /> public FileStatus getFileInfo(String src) throws IOException {<br /> return namesystem.getFileInfo(src); // 调用<br /> }</p> <p> /**<br /> * 获取文件系统的统计信息<br /> */<br /> public long[] getStats() throws IOException {<br /> return namesystem.getStats(); // 调用<br /> }</p> <p> /**<br /> * 创建目录<br /> */<br /> public boolean mkdirs(String src, FsPermission masked) throws IOException {<br /> stateChangeLog.debug("*DIR* NameNode.mkdirs: " + src);<br /> if (!checkPathLength(src)) {<br /> throw new IOException("mkdirs: Pathname too long. Limit " + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");<br /> }<br /> return namesystem.mkdirs(src,<br /> new PermissionStatus(UserGroupInformation.getCurrentUGI().getUserName(),<br /> null, masked)); // 调用<br /> }</p> <p> /**<br /> * 设置安全模式<br /> */<br /> public boolean setSafeMode(SafeModeAction action) throws IOException {<br /> return namesystem.setSafeMode(action); // 调用<br /> }</p> <p> /**<br /> * 保存命名空间映像<br /> */<br /> public void saveNamespace() throws IOException {<br /> namesystem.saveNamespace(); // 调用<br /> }</p> <p> /**<br /> * 刷新Datanode结点的列表<br /> */<br /> public void refreshNodes() throws IOException {<br /> namesystem.refreshNodes(new Configuration()); // 调用<br /> }</p> <p> /**<br /> * 完成升级操作（删除在升级期间保存的文件系统的状态，一旦执行该方法，改变将不可逆转）<br /> */<br /> public void finalizeUpgrade() throws IOException {<br /> namesystem.finalizeUpgrade(); // 调用<br /> }</p> <p> /**<br /> * 报告升级进度<br /> */<br /> public UpgradeStatusReport distributedUpgradeProgress(UpgradeAction action) throws IOException {<br /> return namesystem.distributedUpgradeProgress(action); // 调用<br /> }</p> <p> /**<br /> * 保存Namenode的状态信息，写入到指定的文件中<br /> */<br /> public void metaSave(String filename) throws IOException {<br /> namesystem.metaSave(filename); // 调用<br /> }</p> <p> /**<br /> * 设置指定文件或目录的内容摘要信息<br /> */<br /> public ContentSummary getContentSummary(String path) throws IOException {<br /> return namesystem.getContentSummary(path); // 调用<br /> }</p> <p> /**<br /> * 设置指定目录的配额<br /> */<br /> public void setQuota(String path, long namespaceQuota, long diskspaceQuota) throws IOException {<br /> namesystem.setQuota(path, namespaceQuota, diskspaceQuota); // 调用<br /> }</p> <p> /**<br /> * 将指定文件的元数据写入到持久存储中<br /> */<br /> public void fsync(String src, String clientName) throws IOException {<br /> namesystem.fsync(src, clientName); // 调用<br /> }</p> <p> /**<br /> * 设置文件修改时间<br /> */<br /> public void setTimes(String src, long mtime, long atime) throws IOException {<br /> namesystem.setTimes(src, mtime, atime); // 调用：设置文件修改时间<br /> }

DatanodeProtocol实现

Namenode需要与Datanode进行通信，所以必须实现与DatanodeProtocol协议接口。

下面是对DatanodeProtocol协议接口中定义的基本操作的实现：

/**<br /> * Datanode向Namenode注册，返回Namenode需要验证的有关Datanode的信息DatanodeRegistration的实例<br /> */<br /> public DatanodeRegistration register(DatanodeRegistration nodeReg) throws IOException {<br /> verifyVersion(nodeReg.getVersion()); // 验证协议版本<br /> namesystem.registerDatanode(nodeReg); // 调用：注册指定的Datanode结点<br /> return nodeReg;<br /> }</p> <p> /**<br /> * Datanode向Namenode发送心跳状态报告，显示其当前状态<br /> */<br /> public DatanodeCommand[] sendHeartbeat(DatanodeRegistration nodeReg,<br /> long capacity,<br /> long dfsUsed,<br /> long remaining,<br /> int xmitsInProgress,<br /> int xceiverCount) throws IOException {<br /> verifyRequest(nodeReg);<br /> return namesystem.handleHeartbeat(nodeReg, capacity, dfsUsed, remaining, xceiverCount, xmitsInProgress); // 调用：处理Datanode发送的心跳信息<br /> }</p> <p> /**<br /> * 指定Datanode向Namenode报告其上的块状态报告<br /> */<br /> public DatanodeCommand blockReport(DatanodeRegistration nodeReg, long[] blocks) throws IOException {<br /> verifyRequest(nodeReg); // 验证Datanode向Namenode请求的合法性<br /> BlockListAsLongs blist = new BlockListAsLongs(blocks);<br /> stateChangeLog.debug("*BLOCK* NameNode.blockReport: " +"from "+nodeReg.getName()+" "+blist.getNumberOfBlocks() +" blocks");<br /> namesystem.processReport(nodeReg, blist); // 调用：处理块状态报告<br /> if (getFSImage().isUpgradeFinalized())<br /> return DatanodeCommand.FINALIZE;<br /> return null;<br /> }</p> <p> /**<br /> * Datanode向Namenode报告最近接收到的数据块，以及对数据块的操作信息<br /> */<br /> public void blockReceived(DatanodeRegistration nodeReg, Block blocks[], String delHints[]) throws IOException {<br /> verifyRequest(nodeReg); // 验证请求是否合法<br /> stateChangeLog.debug("*BLOCK* NameNode.blockReceived: " +"from "+nodeReg.getName()+" "+blocks.length+" blocks.");<br /> for (int i = 0; i < blocks.length; i++) {<br /> namesystem.blockReceived(nodeReg, blocks[i], delHints[i]); // 调用<br /> }<br /> }</p> <p> /**<br /> * 报告错误信息<br /> */<br /> public void errorReport(DatanodeRegistration nodeReg, int errorCode, String msg) throws IOException {<br /> String dnName = (nodeReg == null ? "unknown DataNode" : nodeReg.getName());<br /> LOG.info("Error report from " + dnName + ": " + msg); // 将来自指定的Datanode的出错信息登录到日志文件中<br /> if (errorCode == DatanodeProtocol.NOTIFY) {<br /> return;<br /> }<br /> verifyRequest(nodeReg);<br /> if (errorCode == DatanodeProtocol.DISK_ERROR) { // 如果是对应的Datanode出现了磁盘错误<br /> namesystem.removeDatanode(nodeReg); // 调用：删除该Datanode的描述信息<br /> }<br /> }</p> <p> /**<br /> * 在分布式系统升级过程中，向Namenode发送命令<br /> */<br /> public UpgradeCommand processUpgradeCommand(UpgradeCommand comm) throws IOException {<br /> return namesystem.processDistributedUpgradeCommand(comm); // 调用<br /> }

其它实现

用于Namenode与客户端通信，还定义了如下几个与Namenode管理文件系统命名空间相关的操作，主要是对EditLog和FsImage文件的操作：

/**<br /> * 当前EditLog日志文件的大小<br /> */<br /> public long getEditLogSize() throws IOException {<br /> return namesystem.getEditLogSize(); // 调用<br /> }</p> <p> /**<br /> * 回滚EditLog文件<br /> */<br /> public CheckpointSignature rollEditLog() throws IOException {<br /> return namesystem.rollEditLog(); // 调用<br /> }</p> <p> /**<br /> * 回滚FsImage映像文件<br /> */<br /> public void rollFsImage() throws IOException {<br /> namesystem.rollFSImage(); // 调用<br /> }</p> <p> /**<br /> * 获取FsImage映像文件名称<br /> */<br /> public File getFsImageName() throws IOException {<br /> return getFSImage().getFsImageName();<br /> }</p> <p> /**<br /> * 获取FsImage映像文件<br /> */<br /> public FSImage getFSImage() {<br /> return namesystem.dir.fsImage;<br /> }</p> <p> /**<br /> * 获取到FsImage映像文件的所有检查点<br /> */<br /> public File[] getFsImageNameCheckpoint() throws IOException {<br /> return getFSImage().getFsImageNameCheckpoint();<br /> }

在Namenode上，对文件系统名字空间进行的任何操作，Hadoop采用记录事务日志的方式来保存这些操作的记录，对应的事务日志文件就是EditLog文件，该文件存放在磁盘上。而FsImage映像文件，是Hadoop集群工作的过程中使用的，便于Namenode进程管理文件系统的命名空间。当Namenode启动的时候，会首先读取EditLog与FsImage文件，并将EditLog作用于FsImage，因为此时FsImage文件一定是最新的事务日志记录文件的映像，并将EditLog文件删除（因为它可能会在Hadoop集群工作过程中变成旧的文件，也就是不是最新的事务记录）；在Namenode启动之后工作的过程中，仍然会有对文件系统命名空间执行操作的事务记录，这些记录都被保存到一个新的EditLog事务日志文件中，以便下次启动Namenode的时候，将最新的事务日志记录作用于FsImage上。

另外，Namenode类中实现了创建一个Namenode实例的方法createNameNode，该方法是static的，通过命令行参数一个Hadoop配置类实例来创建一个Namenode实例，方法实现如下所示：

public static NameNode createNameNode(String argv[], Configuration conf) throws IOException {<br /> if (conf == null)<br /> conf = new Configuration(); // 如果指定的Configuration类实例为null，创一个默认的配置类实例<br /> StartupOption startOpt = parseArguments(argv); // 解析命令行获取到Namenode启动参数<br /> if (startOpt == null) {<br /> printUsage(); // 如果解析命令行参数返回null，打印命令用法信息<br /> return null;<br /> }<br /> setStartupOption(conf, startOpt); // 设置Namenode启动选项</p> <p> switch (startOpt) { // 根据设置的参数选项，进行处理<br /> case FORMAT:<br /> boolean aborted = format(conf, true);<br /> System.exit(aborted ? 1 : 0);<br /> case FINALIZE:<br /> aborted = finalize(conf, true);<br /> System.exit(aborted ? 1 : 0);<br /> default:<br /> }</p> <p> NameNode namenode = new NameNode(conf); // 构造一个Namenode实例<br /> return namenode;<br /> }

关于配置启动选项，可以查看org.apache.hadoop.hdfs.server.common.HdfsConstants接口中定义了枚举类StartupOption，如下所示：

static public enum StartupOption{<br /> FORMAT ("-format"), // 格式化HDFS选项<br /> REGULAR ("-regular"), // 正常启动选项<br /> UPGRADE ("-upgrade"), // 分布式升级选项<br /> ROLLBACK("-rollback"), // 事务回滚选项<br /> FINALIZE("-finalize"), // 确认升级完成选项<br /> IMPORT ("-importCheckpoint"); //导入检查点选项</p> <p> private String name = null;<br /> private StartupOption(String arg) {this.name = arg;}<br /> public String getName() {return name;}<br /> }

上面创建一个Namenode，在解析完命令行参数后，得到一个枚举类实例，然后根据它来设置Namenode启动选项，如下所示：

private static void setStartupOption(Configuration conf, StartupOption opt) {<br /> conf.set("dfs.namenode.startup", opt.toString()); // 实际上，将命令行解析得到的启动选项，设置到一个Confirmation配置实例上<br /> }

接着，在构造Namenode实例之前，根据设置的附加启动选项，做一个预处理（格式化或某次升级之后的确认动作），如果是格式化操作，具体操作为：

/**<br /> * 验证配置的目录是否存在，如果验证合法则执行格式化操作<br /> */<br /> private static boolean format(Configuration conf, boolean isConfirmationNeeded) throws IOException {<br /> Collection<File> dirsToFormat = FSNamesystem.getNamespaceDirs(conf); // 通过配置实例conf获取到需要进行格式化的文件系统命名空间目录<br /> Collection<File> editDirsToFormat = FSNamesystem.getNamespaceEditsDirs(conf); // 获取到需要进行格式化的文件系统命名空间修改目录<br /> for(Iterator<File> it = dirsToFormat.iterator(); it.hasNext();) { // 验证上述目录的合法性<br /> File curDir = it.next();<br /> if (!curDir.exists())<br /> continue;<br /> if (isConfirmationNeeded) {<br /> System.err.print("Re-format filesystem in " + curDir +" ? (Y or N) ");<br /> if (!(System.in.read() == 'Y')) {<br /> System.err.println("Format aborted in "+ curDir);<br /> return true; // 放弃对文件系统命名空间的格式化操作<br /> }<br /> while(System.in.read() != '/n'); // discard the enter-key<br /> }<br /> }</p> <p> FSNamesystem nsys = new FSNamesystem(new FSImage(dirsToFormat, editDirsToFormat), conf); // 基于构造的FsImage文件实例来创建文件系统命名空间<br /> nsys.dir.fsImage.format(); // 对FsImage文件进行格式化<br /> return false;<br /> }

关于FsImage与EditLog在文件系统中的所在位置的选择，在org.apache.hadoop.hdfs.server.namenode.FSNamesystem类中可以看到详细的情况。

如果是确认升级完成命令行，则实现如下所示：

private static boolean finalize(Configuration conf, boolean isConfirmationNeeded) throws IOException {<br /> Collection<File> dirsToFormat = FSNamesystem.getNamespaceDirs(conf); // 文件系统命名空间文件FsImage所在目录<br /> Collection<File> editDirsToFormat = FSNamesystem.getNamespaceEditsDirs(conf); // EditLog所在目录<br /> FSNamesystem nsys = new FSNamesystem(new FSImage(dirsToFormat, editDirsToFormat), conf); // 构造FSNamesystem<br /> System.err.print(<br /> "/"finalize/" will remove the previous state of the files system./n"<br /> + "Recent upgrade will become permanent./n"<br /> + "Rollback option will not be available anymore./n");<br /> if (isConfirmationNeeded) {<br /> System.err.print("Finalize filesystem state ? (Y or N) ");<br /> if (!(System.in.read() == 'Y')) {<br /> System.err.println("Finalize aborted.");<br /> return true; // 放弃升级确认<br /> }<br /> while(System.in.read() != '/n'); // discard the enter-key<br /> }<br /> nsys.dir.fsImage.finalizeUpgrade(); // 调用FSImage类的finalizeUpgrade方法实现升级完成的确认<br /> return false;<br /> }

关于FsImage文件的操作，可以查看org.apache.hadoop.hdfs.server.namenode.FSImage类的具体实现。在后面会专门对FSImage类进行详细阅读分析的。

下面，对Namenode类的实现做一个总结：

从Namenode实现类来看，它主要是实现了Namenode服务器进程与Datanode进程、文件系统客户端进程，以及Secondary NameNode进程进行交互过程中一些重要的基本的操作，具体表现为，Namenode实现了ClientProtocol协议来与客户端交互，实现了DatanodeProtocol协议来与Datanode进行交互，实现了NamenodeProtocol协议来与Secondary NameNode进行交互。而且，该类还给出了一个static方法，通过命令行的方式用来构造一个Namenode实例，并启动Namenode服务器进程。