hadoop 源码分析（二）HDFS nameNode 之 FSNamesystem初始化源码分析之加载fsImage 和 edits log
上一篇 讲解了nameNode启动的时候，NameNodeHttpServer的启动流程，其实简单来说就是基于hadoop自己实现的HttpServer2服务绑定一个InetSokcetAddress地址，也就是端口号，端口号哪来的？默认配置文件获取呗，最后在将HttpServer2中绑定一些servlet来处理url请求就完成了我们50070端口的请求处理。
那么本篇分析nameNode进程第二个比较核心的，应该说是最核心的组件 FSNamesystem，为什么说是最核心组件呢，因为元数据管理和block的管理都在这个里面进行操作。
直接进入正题，先回顾一下fsimage（全量快照） 文件 以及 edits log（增量事务记录） 文件，自己回忆下这两个文件的重要性。
找到FSNamesystem 的入口
找到之前的nameNode初始化的地方，里面的loadNamesystem（conf）方法，这个就是核心组件FSNamesystem 构建的入口
/** * Initialize name-node. * * @param conf the configuration */ protected void initialize(Configuration conf) throws IOException { //其实这些配置 就是对应着我们配置的 hdfs-site.xml 或者 core-default.xml或者其他文件中一些配置信息 //这些一般也不是很重要 if (conf.get(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS) == null) { String intervals = conf.get(DFS_METRICS_PERCENTILES_INTERVALS_KEY); if (intervals != null) { conf.set(HADOOP_USER_GROUP_METRICS_PERCENTILES_INTERVALS, intervals); } } UserGroupInformation.setConfiguration(conf); loginAsNameNodeUser(conf); NameNode.initMetrics(conf, this.getRole()); StartupProgressMetrics.register(startupProgress); //如果是nameNode 启动一个httpServer if (NamenodeRole.NAMENODE == role) { startHttpServer(conf); } this.spanReceiverHost = SpanReceiverHost.getInstance(conf); // 初始化FSNameSystem 核心组件 loadNamesystem(conf); //初始化rpc server 组件 rpcServer = createRpcServer(conf); if (clientNamenodeAddress == null) { // This is expected for MiniDFSCluster. Set it now using // the RPC server's bind address. clientNamenodeAddress = NetUtils.getHostPortString(rpcServer.getRpcAddress()); LOG.info("Clients are to use " clientNamenodeAddress " to access" " this namenode/service."); } //如果是NameNode 设置NameNodeAddress 以及 FsImage if (NamenodeRole.NAMENODE == role) { httpServer.setNameNodeAddress(getNameNodeAddress()); httpServer.setFSImage(getFSImage()); } pauseMonitor = new JvmPauseMonitor(conf); pauseMonitor.start(); metrics.getJvmMetrics().setPauseMonitor(pauseMonitor); //一些公共服务的初始化 startCommonServices(conf); }这里面会做一些什么操作呢，跟进去看一下
protected void loadNamesystem(Configuration conf) throws IOException { //创建和初始化FSNameSystem //nameNode启动的时候 会讲磁盘上的fsimange 文件 以及 edits log 文件读取到内存中进行合并， // 合并形成最新的元数据 //loadFromDisk 就是讲fsimage 和 edits log 文件读取到内存进行合并 //合并成最新的元数据 就会作为FSNamesystem存放在内存中 this.namesystem = FSNamesystem.loadFromDisk(conf); }还是原先那样，如果有注释，尽可能的把注释看懂，并且结合自己强大的技术底蕴，大胆的猜测里面做了什么操作。。。
/** * Instantiates an FSNamesystem loaded from the image and edits * directories specified in the passed Configuration. * 实例化一个FSNamesystem，怎么实例化呢？加载配置文件中指定的 image 和 edits 文件目录 * 如果没有指定呢，hadoop肯定自己默认的文件目录 * * @param conf the Configuration which specifies the storage directories * from which to load * 包含了存储目录的Configuration，虽然有默认的，至少也要传递过来把，对吧，肯定是通过可识别方式，难不成用眼神？ * * @return an FSNamesystem which contains the loaded namespace * 返回一个包含了加载了 namespace 元数据的 FSNamesystem实例 * * @throws IOException if loading fails */ static FSNamesystem loadFromDisk(Configuration conf) throws IOException { checkConfiguration(conf); //构建FSImage，怎么构建？肯定是从磁盘加载 //FSImage 就是一个时间点的元数据快照信息，其实也就是元数据信息 //FSNamesystem.getNamespaceDirs(conf) 获取元数据的目录，肯定是找指定的配置信息，如果没有肯定就是默认值啦 // 默认值在hdfs-default.xml 以及 hadoop common 中的core-default.xml 文件中 //file://${hadoop.tmp.dir}/dfs/name ${hadoop.tmp.dir}：/tmp/hadoop-${user.name} //可以自己观察下启动的namenode进程的这个目录是否和这个匹配 FSImage fsImage = new FSImage(conf, FSNamesystem.getNamespaceDirs(conf), //FSNamesystem.getNamespaceEditsDirs(conf)) 获取edits log 的目录，和上面的一样对不对 //默认情况下 edis log 和namespace 是在同一个目录下，可以进去看下配置信息 FSNamesystem.getNamespaceEditsDirs(conf)); //实例化 FSnamesystem 对象，讲fsImage对象放入到了FSNamesystem中 FSNamesystem namesystem = new FSNamesystem(conf, fsImage, false); StartupOption startOpt = NameNode.getStartupOption(conf); if (startOpt == StartupOption.RECOVER) { namesystem.setSafeMode(SafeModeAction.SAFEMODE_ENTER); } long loadStart = now(); try { //这里就是说通过FSNamesystem 将fsimage 以及 edits log 加载到内存中 //然后再内存中合并两个文件，形成新的fsImange 信息 //（注：默认情况下 每隔一段时间 就会有check point 将旧的fsimage 与 edits log // 就行合并形成新的fsimage文件，启动的时候肯定也需要合并才能形成新的fsimage文件 对吧） //最后再内存中持有一份完整的元数据信息 namesystem.loadFSImage(startOpt); } catch (IOException ioe) { LOG.warn("Encountered exception loading fsimage", ioe); fsImage.close(); throw ioe; } long timeTakenToLoadFSImage = now() - loadStart; LOG.info("Finished loading FSImage in " timeTakenToLoadFSImage " msecs"); NameNodeMetrics nnMetrics = NameNode.getNameNodeMetrics(); if (nnMetrics != null) { nnMetrics.setFsImageLoadTime((int) timeTakenToLoadFSImage); } return namesystem; }大家可以看到里面做了什么操作呢？
1.构建了一个fsImage对象，注意传入的参数名称，可以发现传入了fsImage 以及edits log，这里大概就猜测出新的fsImage 对象肯定是通过两个合并成新的，但是这里并没有加载数据
2.将新的完整的fsImage元数据信息传递给了FSNamesystem
3.最后加载，意思就是前面只是在初始化一些环境以及配置选项信息以及检查，到这里loadFromDisk才是真正的加载合并
构建FSNamesystem其实里面就是获取一些配置信息以及初始化一些选项，我们简单看一下，里面的内容很多我们主要看下注释 ，看到没，看到没
注释明确说了，不会加载数据，只是构建，如果加载数据使用loadFromDisk，所以注意上面的关联
/** * Create an FSNamesystem associated with the specified image. * 通过指定的 image 元数据快照创建一个 FSNamesystem对象 * Note that this does not load any data off of disk -- if you would * like that behavior, use {@link #loadFromDisk(Configuration)} * 注意 这里不从磁盘加载任何的数据，如果想加载数据，请使用 loadFromDisk(Configuration) * 看到没，这里明确说了不加载数据对吧 * @param conf configuration * @param fsImage The FSImage to associate with * @param ignoreRetryCache Whether or not should ignore the retry cache setup * step. For Secondary NN this should be set to true. * @throws IOException on bad configuration */ FSNamesystem(Configuration conf, FSImage fsImage, boolean ignoreRetryCache) throws IOException {那么我们就直接看下核心的加载数据的方法
loadFSImage（）方法里面大概会有哪些操作？
1.肯定就是加载 fsImage 文件 和 edits log 文件 进行合并为一份完整的内存元数据信息fsImage，
2.将新的fsImage 文件进行覆盖
3.那么edits log 文件呢，肯定也会重新刷新一份。
/*** 肯定就是加载 fsImage 文件 和 edits log 文件 进行合并为一份完整的内存元数据信息，在进行* 磁盘回写，同时注意哦，edits log 文件肯定也要重新来一份新的**/private void loadFSImage(StartupOption startOpt) throws IOException { final FSImage fsImage = getFSImage(); //刚启动的时候 namenode 读取fsimage 以及 edits log 两个文件 //然后再内存中合并形成一个新的fsiamge文件 包含了完整的最新的元数据信息 //然后重新存储在磁盘，替换旧的fsimage 文件，同时会打开一个edits log 文件， // format before starting up if requested //处理下数据， if (startOpt == StartupOption.FORMAT) { fsImage.format(this, fsImage.getStorage().determineClusterId());// reuse current id startOpt = StartupOption.REGULAR; } boolean success = false; writeLock(); try { // We shouldn't be calling saveNamespace if we've come up in standby state. //不会存储元数据，如果我们启动了一个 standby 服务 MetaRecoveryContext recovery = startOpt.createRecoveryContext(); //进行数据加载和 合并 final boolean staleImage = fsImage.recoverTransitionRead(startOpt, this, recovery); if (RollingUpgradeStartupOption.ROLLBACK.matches(startOpt) || RollingUpgradeStartupOption.DOWNGRADE.matches(startOpt)) { rollingUpgradeInfo = null; } final boolean needToSave = staleImage && !haEnabled && !isRollingUpgrade(); LOG.info("Need to save fs image? " needToSave " (staleImage=" staleImage ", haEnabled=" haEnabled ", isRollingUpgrade=" isRollingUpgrade() ")"); //如果需要将fsImage 存储到磁盘 就调用 //其实里面就是将最新的 if (needToSave) { fsImage.saveNamespace(this); } else { updateStorageVersionForRollingUpgrade(fsImage.getLayoutVersion(), startOpt); // No need to save, so mark the phase done. StartupProgress prog = NameNode.getStartupProgress(); prog.beginPhase(Phase.SAVING_CHECKPOINT); prog.endPhase(Phase.SAVING_CHECKPOINT); } //打开一个新的edits log 文件 去进行写入 // This will start a new log segment and write to the seen_txid file, so // we shouldn't do it when coming up in standby state if (!haEnabled || (haEnabled && startOpt == StartupOption.UPGRADE) || (haEnabled && startOpt == StartupOption.UPGRADEONLY)) { fsImage.openEditLogForWrite(); } success = true; } finally { if (!success) { fsImage.close(); } writeUnlock(); } imageLoadComplete(); }我们可以看到除了一些常规的check 和 formate 那么比较重要的就是加载fsImage 以及 edits log 文件的方法入口
fsImage.recoverTransitionRead(startOpt, this, recovery);
大致想想里面会做什么操作？就是将fsImage 文件 和 edits log 文件加载并合并
/** * Analyze storage directories. * Recover from previous transitions if required. * Perform fs state transition if necessary depending on the namespace info. * Read storage info. * 分析存储的目录，什么存储目录，就是存储fsImage 以及edits log 的目录、 * 从以前的状态恢复 * 根据元信息 判断是否执行fs状态的转换 * 读取存储的信息 * 上面意思是什么呢，大概意思就是如果以前有fsImage 和 edits log 就从文件信息中加载出来 并进行恢复 * @throws IOException * @return true if the image needs to be saved or false otherwise */ boolean recoverTransitionRead(StartupOption startOpt, FSNamesystem target, MetaRecoveryContext recovery) throws IOException { assert startOpt != StartupOption.FORMAT : "NameNode formatting should be performed before reading the image"; //获取fsImage 文件资源地址 其实也就是目录 Collection<URI> imageDirs = storage.getImageDirectories(); //获取edits log 目录 Collection<URI> editsDirs = editLog.getEditURIs(); // none of the data dirs exist if((imageDirs.size() == 0 || editsDirs.size() == 0) && startOpt != StartupOption.IMPORT) throw new IOException( "All specified directories are not accessible or do not exist."); // 1. For each data directory calculate its state and // check whether all is consistent before transitioning. //检查每个数据目录，判断是否状态一致性，什么意思呢， // 进行数据恢复 里面就是对一些之前停机的时候 更新 回滚 新增数据的恢复操作 Map<StorageDirectory, StorageState> dataDirStates = new HashMap<StorageDirectory, StorageState>(); boolean isFormatted = recoverStorageDirs(startOpt, storage, dataDirStates); if (LOG.isTraceEnabled()) { LOG.trace("Data dir states:\n " Joiner.on("\n ").withKeyValueSeparator(": ") .join(dataDirStates)); } if (!isFormatted && startOpt != StartupOption.ROLLBACK && startOpt != StartupOption.IMPORT) { throw new IOException("NameNode is not formatted."); } int layoutVersion = storage.getLayoutVersion(); if (startOpt == StartupOption.METADATAVERSION) { System.out.println("HDFS Image Version: " layoutVersion); System.out.println("Software format version: " HdfsConstants.NAMENODE_LAYOUT_VERSION); return false; } if (layoutVersion < Storage.LAST_PRE_UPGRADE_LAYOUT_VERSION) { NNStorage.checkVersionUpgradable(storage.getLayoutVersion()); } if (startOpt != StartupOption.UPGRADE && startOpt != StartupOption.UPGRADEONLY && !RollingUpgradeStartupOption.STARTED.matches(startOpt) && layoutVersion < Storage.LAST_PRE_UPGRADE_LAYOUT_VERSION && layoutVersion != HdfsConstants.NAMENODE_LAYOUT_VERSION) { throw new IOException( "\nFile system image contains an old layout version " storage.getLayoutVersion() ".\nAn upgrade to version " HdfsConstants.NAMENODE_LAYOUT_VERSION " is required.\n" "Please restart NameNode with the \"" RollingUpgradeStartupOption.STARTED.getOptionString() "\" option if a rolling upgrade is already started;" " or restart NameNode with the \"" StartupOption.UPGRADE.getName() "\" option to start" " a new upgrade."); } //执行一些启动选项以及一些二更操作 storage.processStartupOptionsForUpgrade(startOpt, layoutVersion); // 2. Format unformatted dirs. for (Iterator<StorageDirectory> it = storage.dirIterator(); it.hasNext();) { StorageDirectory sd = it.next(); StorageState curState = dataDirStates.get(sd); switch(curState) { case NON_EXISTENT: throw new IOException(StorageState.NON_EXISTENT " state cannot be here"); case NOT_FORMATTED: LOG.info("Storage directory " sd.getRoot() " is not formatted."); LOG.info("Formatting ..."); sd.clearDirectory(); // create empty currrent dir break; default: break; } } // 3. Do transitions switch(startOpt) { case UPGRADE: case UPGRADEONLY: doUpgrade(target); return false; // upgrade saved image already case IMPORT: doImportCheckpoint(target); return false; // import checkpoint saved image already case ROLLBACK: throw new AssertionError("Rollback is now a standalone command, " "NameNode should not be starting with this option."); case REGULAR: default: // just load the image } //真正的加载fsImage 和 edits log 文件进行合并 return loadFSImage(target, startOpt, recovery); }我们其实又看到了一个loadFSImage 操作，这个里面会是真的加载 数据进行合并？
/** * Choose latest image from one of the directories, * load it and merge with the edits. * * 选择最新的 image 全量快照 和 edits log 文件进行合并 * 哇，终于看到了 * * Saving and loading fsimage should never trigger symlink resolution. * The paths that are persisted do not have *intermediate* symlinks * because intermediate symlinks are resolved at the time files, * directories, and symlinks are created. All paths accessed while * loading or saving fsimage should therefore only see symlinks as * the final path component, and the functions called below do not * resolve symlinks that are the final path component. * * @return whether the image should be saved * @throws IOException */ private boolean loadFSImage(FSNamesystem target, StartupOption startOpt, MetaRecoveryContext recovery) throws IOException { final boolean rollingRollback = RollingUpgradeStartupOption.ROLLBACK.matches(startOpt); final EnumSet<NameNodeFile> nnfs; if (rollingRollback) { // if it is rollback of rolling upgrade, only load from the rollback image nnfs = EnumSet.of(NameNodeFile.IMAGE_ROLLBACK); } else { // otherwise we can load from both IMAGE and IMAGE_ROLLBACK nnfs = EnumSet.of(NameNodeFile.IMAGE, NameNodeFile.IMAGE_ROLLBACK); } final FSImageStorageInspector inspector = storage .readAndInspectDirs(nnfs, startOpt); isUpgradeFinalized = inspector.isUpgradeFinalized(); List<FSImageFile> imageFiles = inspector.getLatestImages(); StartupProgress prog = NameNode.getStartupProgress(); prog.beginPhase(Phase.LOADING_FSIMAGE); File phaseFile = imageFiles.get(0).getFile(); prog.setFile(Phase.LOADING_FSIMAGE, phaseFile.getAbsolutePath()); prog.setSize(Phase.LOADING_FSIMAGE, phaseFile.length()); boolean needToSave = inspector.needToSave(); Iterable<EditLogInputStream> editStreams = null; initEditLog(startOpt); if (NameNodeLayoutVersion.supports( LayoutVersion.Feature.TXID_BASED_LAYOUT, getLayoutVersion())) { // If we're open for write, we're either non-HA or we're the active NN, so // we better be able to load all the edits. If we're the standby NN, it's // OK to not be able to read all of edits right now. // In the meanwhile, for HA upgrade, we will still write editlog thus need // this toAtLeastTxId to be set to the max-seen txid // For rollback in rolling upgrade, we need to set the toAtLeastTxId to // the txid right before the upgrade marker. long toAtLeastTxId = editLog.isOpenForWrite() ? inspector .getMaxSeenTxId() : 0; if (rollingRollback) { // note that the first image in imageFiles is the special checkpoint // for the rolling upgrade toAtLeastTxId = imageFiles.get(0).getCheckpointTxId() 2; } ///加载edits log 文件 editStreams = editLog.selectInputStreams( imageFiles.get(0).getCheckpointTxId() 1, toAtLeastTxId, recovery, false); } else { editStreams = FSImagePreTransactionalStorageInspector .getEditLogStreams(storage); } int maxOpSize = conf.getInt(DFSConfigKeys.DFS_NAMENODE_MAX_OP_SIZE_KEY, DFSConfigKeys.DFS_NAMENODE_MAX_OP_SIZE_DEFAULT); for (EditLogInputStream elis : editStreams) { elis.setMaxOpSize(maxOpSize); } for (EditLogInputStream l : editStreams) { LOG.debug("Planning to load edit log stream: " l); } if (!editStreams.iterator().hasNext()) { LOG.info("No edit log streams selected."); } FSImageFile imageFile = null; for (int i = 0; i < imageFiles.size(); i ) { try { imageFile = imageFiles.get(i); //记载Image文件 loadFSImageFile(target, recovery, imageFile, startOpt); break; } catch (IOException ioe) { LOG.error("Failed to load image from " imageFile, ioe); target.clear(); imageFile = null; } } // Failed to load any images, error out if (imageFile == null) { FSEditLog.closeAllStreams(editStreams); throw new IOException("Failed to load an FSImage file!"); } prog.endPhase(Phase.LOADING_FSIMAGE); if (!rollingRollback) { long txnsAdvanced = loadEdits(editStreams, target, startOpt, recovery); needToSave |= needsResaveBasedOnStaleCheckpoint(imageFile.getFile(), txnsAdvanced); if (RollingUpgradeStartupOption.DOWNGRADE.matches(startOpt)) { // rename rollback image if it is downgrade renameCheckpoint(NameNodeFile.IMAGE_ROLLBACK, NameNodeFile.IMAGE); } } else { // Trigger the rollback for rolling upgrade. Here lastAppliedTxId equals // to the last txid in rollback fsimage. rollingRollback(lastAppliedTxId 1, imageFiles.get(0).getCheckpointTxId()); needToSave = false; } editLog.setNextTxId(lastAppliedTxId 1); return needToSave; } /** rollback for rolling upgrade. */ private void rollingRollback(long discardSegmentTxId, long ckptId) throws IOException { // discard discard unnecessary editlog segments starting from the given id this.editLog.discardSegments(discardSegmentTxId); // rename the special checkpoint renameCheckpoint(ckptId, NameNodeFile.IMAGE_ROLLBACK, NameNodeFile.IMAGE, true); // purge all the checkpoints after the marker archivalManager.purgeCheckpoinsAfter(NameNodeFile.IMAGE, ckptId); String nameserviceId = DFSUtil.getNamenodeNameServiceId(conf); if (HAUtil.isHAEnabled(conf, nameserviceId)) { // close the editlog since it is currently open for write this.editLog.close(); // reopen the editlog for read this.editLog.initSharedJournalsForRead(); } }进行loadFSImanageFile文件的加载，这里就比较枯燥了，为什么？因为我们明显要有大局观，首先我们肯定直到一定是加载文件，对吧？那么文件最后加载到哪里去进行存储？因为我们的FSNamesystem 是核心管理元数据的组件，大家肯定就也想到了，最终数据肯定是加载到其中啦，所以大家注意看这里对啊不，一直有将FSNamesystem 作为参数再进行传递
/** * 这里面就不仔细去详细的跟到文件加载了， * loadFSImage（）方法就是最终加载文件的方法 */ void loadFSImageFile(FSNamesystem target, MetaRecoveryContext recovery, FSImageFile imageFile, StartupOption startupOption) throws IOException { LOG.debug("Planning to load image :\n" imageFile); StorageDirectory sdForProperties = imageFile.sd; storage.readProperties(sdForProperties, startupOption); if (NameNodeLayoutVersion.supports( LayoutVersion.Feature.TXID_BASED_LAYOUT, getLayoutVersion())) { // For txid-based layout, we should have a .md5 file // next to the image file boolean isRollingRollback = RollingUpgradeStartupOption.ROLLBACK .matches(startupOption); loadFSImage(imageFile.getFile(), target, recovery, isRollingRollback); } else if (NameNodeLayoutVersion.supports( LayoutVersion.Feature.FSIMAGE_CHECKSUM, getLayoutVersion())) { // In 0.22, we have the checksum stored in the VERSION file. String md5 = storage.getDeprecatedProperty( NNStorage.DEPRECATED_MESSAGE_DIGEST_PROPERTY); if (md5 == null) { throw new InconsistentFSStateException(sdForProperties.getRoot(), "Message digest property " NNStorage.DEPRECATED_MESSAGE_DIGEST_PROPERTY " not set for storage directory " sdForProperties.getRoot()); } loadFSImage(imageFile.getFile(), new MD5Hash(md5), target, recovery, false); } else { // We don't have any record of the md5sum loadFSImage(imageFile.getFile(), null, target, recovery, false); } }/** * Load in the filesystem image from file. It's a big list of * filenames and blocks. * 加载 fsimage文件， */ private void loadFSImage(File curFile, MD5Hash expectedMd5, FSNamesystem target, MetaRecoveryContext recovery, boolean requireSameLayoutVersion) throws IOException { // BlockPoolId is required when the FsImageLoader loads the rolling upgrade // information. Make sure the ID is properly set. target.setBlockPoolId(this.getBlockPoolID()); //一个持有FSNamesystem 以及 conf 对象的loader，加载器嘛，为什么持有FSNamesystem？ //因为FSNamesystem 是管理元数据的核心组件，最终的元数据都是存储在FSNamesystem 中对把 FSImageFormat.LoaderDelegator loader = FSImageFormat.newLoader(conf, target); loader.load(curFile, requireSameLayoutVersion); // Check that the image digest we loaded matches up with what // we expected MD5Hash readImageMd5 = loader.getLoadedImageMd5(); if (expectedMd5 != null && !expectedMd5.equals(readImageMd5)) { throw new IOException("Image file " curFile " is corrupt with MD5 checksum of " readImageMd5 " but expecting " expectedMd5); } long txId = loader.getLoadedImageTxId(); LOG.info("Loaded image for txid " txId " from " curFile); lastAppliedTxId = txId; storage.setMostRecentCheckpointInfo(txId, curFile.lastModified()); }其实知道这大家也就知道了其实就是实例化了一个loader 组件去加载fsImage文件了，底层就是基于文件流加载咯
接下就是加载edits log 以及进行一个合并了入口就是
//执行加载edits log 其实里面也包含了合并 fsImage 与 edits log 的操作 long txnsAdvanced = loadEdits(editStreams, target, startOpt, recovery);这里大家也可以发现其实也是将我们的editStream 目录 以及 FSNamesystem 作为了参数传入，底层说不定也是基于loader 加载edits log 然后放入到FSNamesystem中对吧，当然 肯定也会有一些其他的操作，比如检查，筛选对吧
/** * Load the specified list of edit files into the image. * 加载指定的edit files 文件到image中 * 怎么加载 就是再跑一边记录的指令咯 */ public long loadEdits(Iterable<EditLogInputStream> editStreams, FSNamesystem target) throws IOException { return loadEdits(editStreams, target, null, null); } private long loadEdits(Iterable<EditLogInputStream> editStreams, FSNamesystem target, StartupOption startOpt, MetaRecoveryContext recovery) throws IOException { LOG.debug("About to load edits:\n " Joiner.on("\n ").join(editStreams)); StartupProgress prog = NameNode.getStartupProgress(); prog.beginPhase(Phase.LOADING_EDITS); long prevLastAppliedTxId = lastAppliedTxId; try { //实例化一个loader 加载组件，也是和加载FSImage 文件一样，持有了 FSNamesystem FSEditLogLoader loader = new FSEditLogLoader(target, lastAppliedTxId); // Load latest edits //加载最新的edits log for (EditLogInputStream editIn : editStreams) { LOG.info("Reading " editIn " expecting start txid #" (lastAppliedTxId 1)); try { //进行文件流的加载对吧，底层不用说 肯定是基于文件流咯 loader.loadFSEdits(editIn, lastAppliedTxId 1, startOpt, recovery); } finally { // Update lastAppliedTxId even in case of error, since some ops may // have been successfully applied before the error. lastAppliedTxId = loader.getLastAppliedTxId(); } // If we are in recovery mode, we may have skipped over some txids. if (editIn.getLastTxId() != HdfsConstants.INVALID_TXID) { lastAppliedTxId = editIn.getLastTxId(); } } } finally { FSEditLog.closeAllStreams(editStreams); // update the counts updateCountForQuota(target.dir.rootDir); } prog.endPhase(Phase.LOADING_EDITS); return lastAppliedTxId - prevLastAppliedTxId; }/** * Load an edit log, and apply the changes to the in-memory structure * This is where we apply edits that we've been writing to disk all * along. * 记载edit log，同时将一些更改变化加入到内存结构中，这里其实就是将我们的记录再 * edits log 中的操作指令再进行一次刷新 */ long loadFSEdits(EditLogInputStream edits, long expectedStartingTxId, StartupOption startOpt, MetaRecoveryContext recovery) throws IOException { StartupProgress prog = NameNode.getStartupProgress(); Step step = createStartupProgressStep(edits); prog.beginStep(Phase.LOADING_EDITS, step); fsNamesys.writeLock(); try { long startTime = now(); FSImage.LOG.info("Start loading edits file " edits.getName()); long numEdits = loadEditRecords(edits, false, expectedStartingTxId, startOpt, recovery); FSImage.LOG.info("Edits file " edits.getName() " of size " edits.length() " edits # " numEdits " loaded in " (now()-startTime)/1000 " seconds"); return numEdits; } finally { edits.close(); fsNamesys.writeUnlock(); prog.endStep(Phase.LOADING_EDITS, step); } }到这其实大家也就知道了逻辑了，最后的时候其实在返回一个needToSave ，做什么的我就不说了
小总结：FSNamesystem 初始化的步骤
1.从配置文件中获取fsimage 以及 edits log 目录
2.构建FSnamesystem，并传入 fsimage 以及 edits log 目录以及其他信息
3.loadfromdisk 加载 fsimage 以及edits log 文件到内存
4.将fsimage 和 edits log 文件进行合并成新的fsimage，也就是完整的元数据信息，在内存中持有一份
5.将新的完整的fsimage 文件替换 旧的fsimage 文件
6.开启一个新的edits log 文件记录增量事务
赞 (0)