Hadoop3.2.1 【 HDFS 】源码分析 : 文件系统数据集 [一]

心已赠人 2022-12-20 02:39 133阅读 0赞

### List of articles ###

*  一. 前言
 *  二.Datanode上数据块副本的状态
 *  三.BlockPoolSlice实现
 *   *  3.1.BlockPoolSlice的字段
     *  3.2.BlockPoolSlice的方法
     *   *  3.2.1.构造方法
         *  3.2.1.磁盘空间相关方法
         *  3.2.1.块池目录子文件夹方法
         *  3.2.1.数据块副本操作方法

# 一. 前言 #

Datanode可以配置多个存储目录保存数据块文件（这里要注意， Datanode的多个存储目录存储的数据块并不相同， 并且不同的存储目录可以是异构的， 这样的设计可以提高数据块IO的吞吐率） ， 所以Datanode将底层数据块的管理抽象为多个层次， 并定义不同的类  
来实现各个层次上数据块的管理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

■ BlockPoolSlice： 管理一个指定块池在一个指定存储目录下的所有数据块。 由于Datanode可以定义多个存储目录， 所以块池的数据块会分布在多个存储目录下。一个块池会拥有多个BlockPoolSlice对象， 这个块池对应的所有BlockPoolSlice对象共同管理块池的所有数据块。
    ■ FsVolumelmpl： 管理Datanode一个存储目录下的所有数据块。 由于一个存储目录可以存储多个块池的数据块， 所以FsVolumelmpl会持有这个存储目录中保存的所有块池的BlockPoolSlice对象。
    ■ FsVolumeList： Datanode可以定义多个存储目录， 每个存储目录下的数据块是使用一个FsVolumelmpl对象管理的， 所以Datanode定义了FsVolumeList类保存Datanode上所有的FsVolumelmpl对象， FsVolumeList对FsDatasetlmpl提供类似磁盘的服务。

# 二.Datanode上数据块副本的状态 #

FsDatasetlmpl会持有一个ReplicaMap对象， 维护Datanode上所有数据块副本的状态。

Datanode上保存的数据块副本有下面5种状态。  
■ **FINALIZED**： Datanode上已经完成写操作的副本。 Datanode定义了FinalizedReplica类来保存FINALIZED状态副本的信息。  
■ **RBW（Replica Being Written）** ： 刚刚由HDFS客户端创建的副本或者进行追加写（append） 操作的副本， 副本的数据正在被写入， 部分数据对于客户端是可见的。 Datanode定义了ReplicaBeingWritten类来保存RBW状态副本的信息。  
■ **RUR（Relica Under Recovery）** ： 进行块恢复时的副本。 Datanode定义了ReplicaUnderRecovery类来保存RUR状态副本的信息。  
■ **RWR（ReplicaWaitingToBeRecovered）** ： 如果Datanode重启或者宕机， 所有RBW状态的副本在Datanode重启后都将被加载为RWR状态， RWR会等待块恢复操作。 Datanode定义了ReplicaWaitingToBeRecovered类来保存RWR状态副本的信息。  
■ **TEMPORARY**： Datanode之间复制数据块， 或者进行集群数据块平衡操作（cluster balance） 时， 正在写入副本的状态就是TEMPORARY状态。 和RBW不同的是， **TEMPORARY状态的副本对于客户端是不可见的**， **同时Datanode重启时将会直接删除处于TEMPORARY状态的副本。** Datanode定义了ReplicaInPipeline类来保存TEMPORARY状态副本的信息。

Datanode会将不同状态的副本存储到磁盘的不同目录下， 也就是说， Datanode上副本的状态是会被持久化的。 HDFS使BlockPoolSlice、 FsVolumelmpl以及FsVolumeList类对这些存储目录进行管理

# 三.BlockPoolSlice实现 #

BlockPoolSlice类负责管理指定存储目录下一个指定block pool 的所有数据块。block pool在每个存储目录下都会有一个block pool 目录存储数据块，换句话说， BlockPollSlice管理的就是这个block pool 目录的所有数据块。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]

一个块池目录存在current、finalized、rbw、lazypersist、tmp等目录。  
current目录包含finalized、 rbw以及lazyPersisit三个子目录。  
finalized目录保存了所有FINALIZED状态的副本， rbw目录保存了RBW（正在写） 、 RWR（等待恢复） 、 RUR（恢复中） 状态的副本， tmp目录保存了TEMPORARY状态的副本。  
当在客户端发起写请求而创建一个新的副本时， 这个副本将会被放到rbw目录中。  
当在数据块备份和集群平衡存储过程中创建一个新的副本时， 这个副本就会放到tmp目录中。  
一旦一个副本完成写操作并被提交， 它就会被移动到finalized目录中。  
当Datanode重启时， tmp目录中的所有副本将会被删除， rbw目录中的副本将会被加载为RWR状态， finalized目录中的副本将会被加载为FINALIZED状态。  
lazyPersist目录为HDFS 2.6版本新加入的， 用来支持在内存中写入临时数据块副本以及慢持久化（lazypersist） 数据块副本的特性， lazyPersist目录就是用于保存这些慢持久化的数据块副本的。

finalized子目录的结构比较特殊， 它即包含目录也包含文件。 finalized子目录存储了两种类型的文件。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]

■ 数据块文件， 比如blk\_1073741825数据块文件，1073741825是数据块id。  
■ meta后缀的校验文件， 用来保存数据块文件的校验信息，  
blk\_1073741825\_1001.meta就是校验文件， 文件名中的1001是数据块的版本号

当集群数据量达到一定程度时， finalized目录下的数据块将会非常多。 为了便于组织管理， 数据块将按照数据块id进行散列， 拥有相同散列值的数据块将处于同一个子目录下。  
**在2.7.x版本 finalized目录下最多拥有256个一级子目录， 每个一级子目录下又可以拥有256个二级子目录。**  
**在2.8.x版本以后到现在的3.3.0 finalized目录下最多拥有32个一级子目录， 每个一级子目录下又可以拥有32个二级子目录。**

## 3.1.BlockPoolSlice的字段 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]

■ **bpid**： 记录当前BlockPoolSlice对应的block pool id。  
■ **目录相关字段**： 保存当前块池目录的current子目录（currentDir字段） 、 finalized子目录（finalizedDir字段） 、 lazyPersist子目录（lazypersistDir字段） 、 rbw子目录（rbwDir字段)、 tmp子目录（tmpDir字段） 的引用。  
■ **dfsUsage**： DU类型， 描述当前块池目录的磁盘使用情况。

## 3.2.BlockPoolSlice的方法 ##

### 3.2.1.构造方法 ###

BlockPoolSlice的构造方法首先会创建current目录、 finalized目录、 lazyPersist目录、rbw目录以及tmp目录， 并将这些目录的引用赋值到对应的字段上.完成子目录的创建之后， 构造方法会初始化dfsUsage字段， 用来统计块池目录的磁盘使用情况。 最后构造方法会添加一个钩子， 当Datanode进程结束时保存整个文件系统的磁盘使用信息。

> 调用链\[很长很长…\]:  
> Datanode\#runDatanodeDaemon() ⇒ BlockPoolManager.startAll(); ⇒ BPOfferService\#start() ⇒ BPServiceActor\#start() => BPServiceActor\#run() ⇒ connectToNNAndHandshake(); ==> bpos.verifyAndSetNamespaceInfo(this, nsInfo); ==> DataNode.initBlockPool(this); ⇒ FsDatasetImpl\#addBlockPool() ⇒ FsVolumeList\#addBlockPool() ⇒FsVolumeImpl\#addBlockPool() ==> new BlockPoolSlice(bpid, this, bpdir, c, new Timer());

/**
       * Create a blook pool slice
       * @param bpid Block pool Id
       * @param volume {@link FsVolumeImpl} to which this BlockPool belongs to
       * @param bpDir directory corresponding to the BlockPool
       * @param conf configuration
       * @param timer include methods for getting time
       * @throws IOException
       */
      BlockPoolSlice(String bpid, FsVolumeImpl volume, File bpDir,
          Configuration conf, Timer timer) throws IOException {
        // BP-451827885-192.168.8.156-1584099133244
        this.bpid = bpid;
        // /tools/hadoop-3.2.1/data/hdfs/data
        this.volume = volume;
        //
        this.fileIoProvider = volume.getFileIoProvider();
        //  /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/current
        this.currentDir = new File(bpDir, DataStorage.STORAGE_DIR_CURRENT);
        //  /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/current/finalized
        this.finalizedDir = new File( currentDir, DataStorage.STORAGE_DIR_FINALIZED);
        //  /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/current/lazypersist
        this.lazypersistDir = new File(currentDir, DataStorage.STORAGE_DIR_LAZY_PERSIST);
    
        if (!this.finalizedDir.exists()) {
          if (!this.finalizedDir.mkdirs()) {
            throw new IOException("Failed to mkdirs " + this.finalizedDir);
          }
        }
        // 4096
        this.ioFileBufferSize = DFSUtilClient.getIoFileBufferSize(conf);
        // dfs.datanode.duplicate.replica.deletion : true
        this.deleteDuplicateReplicas = conf.getBoolean(
            DFSConfigKeys.DFS_DATANODE_DUPLICATE_REPLICA_DELETION,
            DFSConfigKeys.DFS_DATANODE_DUPLICATE_REPLICA_DELETION_DEFAULT);
        // dfs.datanode.cached-dfsused.check.interval.ms : 600000 ms  =>  1 minute
        this.cachedDfsUsedCheckTime =
            conf.getLong(
                DFSConfigKeys.DFS_DN_CACHED_DFSUSED_CHECK_INTERVAL_MS,
                DFSConfigKeys.DFS_DN_CACHED_DFSUSED_CHECK_INTERVAL_DEFAULT_MS);
        // ipc.maximum.data.length : 64mb
        this.maxDataLength = conf.getInt(
            CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH,
            CommonConfigurationKeys.IPC_MAXIMUM_DATA_LENGTH_DEFAULT);
    
        this.timer = timer;
    
        // Files that were being written when the datanode was last shutdown
        // are now moved back to the data directory. It is possible that
        // in the future, we might want to do some sort of datanode-local
        // recovery for these blocks. For example, crc validation.
        // tmpDir :  /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/tmp
        this.tmpDir = new File(bpDir, DataStorage.STORAGE_DIR_TMP);
    
        if (tmpDir.exists()) {
          fileIoProvider.fullyDelete(volume, tmpDir);
        }
        // /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/current/rbw
        this.rbwDir = new File(currentDir, DataStorage.STORAGE_DIR_RBW);
    
        // create the rbw and tmp directories if they don't exist.
    
        fileIoProvider.mkdirs(volume, rbwDir);
    
        fileIoProvider.mkdirs(volume, tmpDir);
    
        // Use cached value initially if available. Or the following call will block until the initial du command completes.
        // du -sk /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244
        // 132947968	/tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244
        this.dfsUsage = new CachingGetSpaceUsed.Builder().setPath(bpDir)
                                                         .setConf(conf)
                                                         .setInitialUsed(loadDfsUsed())
                                                         .build();
    
        if (addReplicaThreadPool == null) {
          // initialize add replica fork join pool
          initializeAddReplicaPool(conf);
        }
        // Make the dfs usage to be saved during shutdown.
    
        shutdownHook = new Runnable() {
          @Override
          public void run() {
            if (!dfsUsedSaved) {
              saveDfsUsed();
              addReplicaThreadPool.shutdownNow();
            }
          }
        };
    
        ShutdownHookManager.get().addShutdownHook(shutdownHook,
            SHUTDOWN_HOOK_PRIORITY);
      }

### 3.2.1.磁盘空间相关方法 ###

BlockPoolSlice定义了5个用于操作当前块池目录占用磁盘空间的方法， 包括incDfsUsed()、 decDfsUsed()、 getDfsUsed()、 loadDfsUsed()、 saveDfsUsed()。 这5个方法底层都是通过调用dfsUsage字段的对应方法实现的。  
dfsUsage字段是DU类型的， DU类是用于统计磁盘使用情况的工具类. 是通过shell的方式获取操作系统的磁盘使用情况:

比如: 
    # -k：和--block-size=1k类似，以KB为单位。
    # -s：--summarize，仅列出总量，而不列出每个目录和文件的大小，
    
    du -sk /tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244
    
    返回:
    
    132947968	/tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244

### 3.2.1.块池目录子文件夹方法 ###

我们知道块池目录下会包含若干子文件夹， 包括current、 finalized、 rbw以及lazyPersist等。 BlockPoolSlice定义了4个用于获取这些子文件夹引用的方法， 包括getDirectory()、 getFinalizedDir()、 getLazypersistDir()、 getRbwDir()。 这4个方法的实现非常简单， 直接返回由构造方法初始化的对应字段即可。

File getDirectory() {
        return currentDir.getParentFile();
      }
    
      File getFinalizedDir() {
        return finalizedDir;
      }
    
      File getLazypersistDir() {
        return lazypersistDir;
      }
    
      File getRbwDir() {
        return rbwDir;
      }
    
      File getTmpDir() {
        return tmpDir;
      }

### 3.2.1.数据块副本操作方法 ###

BlockPoolSlice类提供了若干在块池目录下操作数据块副本的方法， 包括创建不同状态的数据块副本、 获取当前块池目录存储副本的状态等， FsVolumelmpl类会通过它持有的BlockPoolSlice对象调用这些方法完成操作数据块副本的功能.

**在块池目录中创建数据块副本**

在当前块池目录中创建副本包括以下几种情况： 创建TEMPORARY状态、 RBW状态以及FINALIZED状态的副本等。  
创建TEMPORARY状态以及RBW状态的副本操作都比较简单， 直接在tmp目录以及rbw目录下通过IO操作创建数据块副本文件即可。

/**
       *
       * 临时文件。
       * 当block完成时，它们被移到finalized块目录中。
       *
       * Temporary files. They get moved to the finalized block directory when
       * the block is finalized.
       */
      File createTmpFile(Block b) throws IOException {
        File f = new File(tmpDir, b.getBlockName());
        //构建tmpFile
        File tmpFile = DatanodeUtil.createFileWithExistsCheck( volume, b, f, fileIoProvider);
        // If any exception during creation, its expected that counter will not be
        // incremented, So no need to decrement
        incrNumBlocks();
        return tmpFile;
      }
    
    
    
      /**
       * RBW files. They get moved to the finalized block directory when
       * the block is finalized.
       */
      File createRbwFile(Block b) throws IOException {
        File f = new File(rbwDir, b.getBlockName());
        File rbwFile = DatanodeUtil.createFileWithExistsCheck(
            volume, b, f, fileIoProvider);
        // If any exception during creation, its expected that counter will not be
        // incremented, So no need to decrement
        incrNumBlocks();
        return rbwFile;
      }

创建FINALIZED状态的副本则由addFinalizedBlock()方法实现， addFinalizedBlock()方法首先调用DatanodeUtil.idToBlockDir()方法获得该副本在finalized目录的存储路径， 然后在这个路径下创建数据块副本文件以及校验文件， 最后更改dfsUsage更新磁盘使用情况

File addFinalizedBlock(Block b, ReplicaInfo replicaInfo) throws IOException {
        //调用DatanodeUtil.idToBlockDir()方法获取副本在finalized目录的存储路径
        File blockDir = DatanodeUtil.idToBlockDir(finalizedDir, b.getBlockId());
        fileIoProvider.mkdirsWithExistsCheck(volume, blockDir);
    
        //在存储路径中创建数据块副本文件以及校验文件
        File blockFile = FsDatasetImpl.moveBlockFiles(b, replicaInfo, blockDir);
        File metaFile = FsDatasetUtil.getMetaFile(blockFile, b.getGenerationStamp());
    
    
        //更新dfsUsage
        if (dfsUsage instanceof CachingGetSpaceUsed) {
          ((CachingGetSpaceUsed) dfsUsage).incDfsUsed(
              b.getNumBytes() + metaFile.length());
        }
        return blockFile;
      }

finalized目录下的副本文件是以多级子目录方式存放的，DatanodeUtil.idToBlockDir()方法提供了确定这个副本存储路径的功能。

/**
       * Get the directory where a finalized block with this ID should be stored.
       * Do not attempt to create the directory.
       * @param root the root directory where finalized blocks are stored
       * @param blockId
       * @return
       */
      public static File idToBlockDir(File root, long blockId) {
    
    
        System.out.println( blockId + " ====>   " + Long.toBinaryString(blockId));
    
        System.out.println(" 0x1F ====> long :   " + 0x1F );
    
        System.out.println(" 0x1F ====>   " + Long.toBinaryString(0x1F));
    
    
    
        // blockld的第三个字节作为一级目录索引
        int d1 = (int) ((blockId >> 16) & 0x1F);
        // blockld的第二个字节作为二级目录索引
    
        System.out.println(" d1 ====>   " + Integer.toBinaryString(d1));
    
        int d2 = (int) ((blockId >> 8) & 0x1F);
    
        System.out.println(" d2 ====>   " + Integer.toBinaryString(d2));
    
        //构造存储路径并返回
        String path = DataStorage.BLOCK_SUBDIR_PREFIX + d1 + SEP +
            DataStorage.BLOCK_SUBDIR_PREFIX + d2;
        return new File(root, path);
      }

**获取块池目录存储副本的状态**  
Datanode在启动时会调用BlockPoolSlice.getVolumeMap()方法获取当前BlockPoolSlice.持有的所有数据块副本的状态， DataNode类会使用一个ReplicaMap 对象保存这些数据块副本的信息以及状态。 这里要注意， **ReplicaMap对象只保存FINALIZED状态以及RBR状态的数据块副本**。 getVolumeMap()的代码如下所示， 它会首先恢复lazyPersist状态的副本， 将它们转变为FINALIZED状态， 然后getVolumeMap()方法会调用addToReplicasMap()方法处理finalized目录以及rbw目录中保存的所有数据块副本。

void getVolumeMap(ReplicaMap volumeMap,
                        final RamDiskReplicaTracker lazyWriteReplicaMap)
          throws IOException {
        // Recover lazy persist replicas, they will be added to the volumeMap
        // when we scan the finalized directory.
        //恢复lazyPersist状态的副本， 先将它们转变为FINALIZED状态
        if (lazypersistDir.exists()) {
          int numRecovered = moveLazyPersistReplicasToFinalized(lazypersistDir);
          FsDatasetImpl.LOG.info(
              "Recovered " + numRecovered + " replicas from " + lazypersistDir);
        }
    
        //将所有FINALIZED状态的副本加入ReplicaMap中
        boolean  success = readReplicasFromCache(volumeMap, lazyWriteReplicaMap);
        if (!success) {
          List<IOException> exceptions = Collections
              .synchronizedList(new ArrayList<IOException>());
          Queue<RecursiveAction> subTaskQueue =
              new ConcurrentLinkedQueue<RecursiveAction>();
    
          // add finalized replicas
          AddReplicaProcessor task = new AddReplicaProcessor(volumeMap,
              finalizedDir, lazyWriteReplicaMap, true, exceptions, subTaskQueue);
          ForkJoinTask<Void> finalizedTask = addReplicaThreadPool.submit(task);
    
          // add rbw replicas
          task = new AddReplicaProcessor(volumeMap, rbwDir, lazyWriteReplicaMap,
              false, exceptions, subTaskQueue);
          ForkJoinTask<Void> rbwTask = addReplicaThreadPool.submit(task);
    
          try {
            finalizedTask.get();
            rbwTask.get();
          } catch (InterruptedException | ExecutionException e) {
            exceptions.add(new IOException(
                "Failed to start sub tasks to add replica in replica map :"
                    + e.getMessage()));
          }
    
          //wait for all the tasks to finish.
          waitForSubTaskToFinish(subTaskQueue, exceptions);
        }
      }
    
    
      private boolean readReplicasFromCache(ReplicaMap volumeMap,  final RamDiskReplicaTracker lazyWriteReplicaMap) {
    
        ReplicaMap tmpReplicaMap = new ReplicaMap(new AutoCloseableLock());
        File replicaFile = new File(currentDir, REPLICA_CACHE_FILE);
    
        // Check whether the file exists or not.
        if (!replicaFile.exists()) {
          LOG.info("Replica Cache file: "+  replicaFile.getPath() +
              " doesn't exist ");
          return false;
        }
    
        long fileLastModifiedTime = replicaFile.lastModified();
    
        if (System.currentTimeMillis() > fileLastModifiedTime + replicaCacheExpiry) {
          LOG.info("Replica Cache file: " + replicaFile.getPath() +
              " has gone stale");
          // Just to make findbugs happy
          if (!replicaFile.delete()) {
            LOG.info("Replica Cache file: " + replicaFile.getPath() +
                " cannot be deleted");
          }
          return false;
        }
        FileInputStream inputStream = null;
        try {
    
          inputStream = fileIoProvider.getFileInputStream(volume, replicaFile);
    
          BlockListAsLongs blocksList =  BlockListAsLongs.readFrom(inputStream, maxDataLength);
    
          if (blocksList == null) {
            return false;
          }
    
          for (BlockReportReplica replica : blocksList) {
            switch (replica.getState()) {
            case FINALIZED:
              //将所有FINALIZED状态的副本加入ReplicaMap中
              addReplicaToReplicasMap(replica, tmpReplicaMap, lazyWriteReplicaMap, true);
              break;
            case RUR:
            case RBW:
            case RWR:
              //将所有RBR状态的副本加入ReplicaMap中
              addReplicaToReplicasMap(replica, tmpReplicaMap, lazyWriteReplicaMap, false);
              break;
            default:
              break;
            }
          }
          // Now it is safe to add the replica into volumeMap
          // In case of any exception during parsing this cache file, fall back
          // to scan all the files on disk.
          for (Iterator<ReplicaInfo> iter =
              tmpReplicaMap.replicas(bpid).iterator(); iter.hasNext(); ) {
            ReplicaInfo info = iter.next();
            // We use a lightweight GSet to store replicaInfo, we need to remove
            // it from one GSet before adding to another.
            iter.remove();
            volumeMap.add(bpid, info);
          }
          LOG.info("Successfully read replica from cache file : "
              + replicaFile.getPath());
          return true;
        } catch (Exception e) {
          // Any exception we need to revert back to read from disk
          // Log the error and return false
          LOG.info("Exception occurred while reading the replicas cache file: "
              + replicaFile.getPath(), e );
          return false;
        }
        finally {
          // close the inputStream
          IOUtils.closeStream(inputStream);
    
          if (!fileIoProvider.delete(volume, replicaFile)) {
            LOG.info("Failed to delete replica cache file: " +
                replicaFile.getPath());
          }
        }
      }

addToReplicasMap()方法会遍历dir参数指定的文件夹下的所有副本文件， 如果是finalized文件夹中的， 则加载为FINALIZED状态； 如果不是finalized文件夹中的， 则判断当前文件夹中是否存在.restart文件（Datanode快速重启时会在磁盘上保存一个文件， 里面存储了一个大概的重启时间， 用于恢复文件的10流） ， 如果存在这个文件， 并且时间还在重启窗口内， 则当前数据块副本可以恢复为RBW状态， 将当前数据块加载为RBW状态； 如果不存在这个文件， 则将当前数据块加载为RWR状态。

/**
       * 将给定目录下的副本添加到卷映射
       * Add replicas under the given directory to the volume map
       * @param volumeMap the replicas map
       * @param dir an input directory
       * @param lazyWriteReplicaMap Map of replicas on transient
       *                                storage.
       * @param isFinalized true if the directory has finalized replicas;
       *                    false if the directory has rbw replicas
       * @param exceptions list of exception which need to return to parent thread.
       * @param subTaskQueue queue of sub tasks
       */
      void addToReplicasMap(ReplicaMap volumeMap, File dir,
          final RamDiskReplicaTracker lazyWriteReplicaMap, boolean isFinalized,
          List<IOException> exceptions, Queue<RecursiveAction> subTaskQueue)
          throws IOException {
        //循环遍历文件夹中的所有文件
        File[] files = fileIoProvider.listFiles(volume, dir);
    
        Arrays.sort(files, FILE_COMPARATOR);
    
        for (int i = 0; i < files.length; i++) {
    
          File file = files[i];
    
          if (file.isDirectory()) {
    
            // Launch new sub task.
            // 启动一个子程序处理
            AddReplicaProcessor subTask = new AddReplicaProcessor(volumeMap, file,
                lazyWriteReplicaMap, isFinalized, exceptions, subTaskQueue);
            subTask.fork();
            subTaskQueue.add(subTask);
          }
    
          //如果文件夹中存在临时文件， 则首先进行恢复操作
          if (isFinalized && FsDatasetUtil.isUnlinkTmpFile(file)) {
            file = recoverTempUnlinkedBlock(file);
            if (file == null) { // the original block still exists, so we cover it
              // in another iteration and can continue here
              // 如果临时文件对应的源文件存在， 那么跳过该临时文件
              continue;
            }
          }
          if (!Block.isBlockFilename(file)) {
            continue;
          }
    
          long genStamp = FsDatasetUtil.getGenerationStampFromFile(
              files, file, i);
    
          long blockId = Block.filename2id(file.getName());
    
          Block block = new Block(blockId, file.length(), genStamp);
    
          addReplicaToReplicasMap(block, volumeMap, lazyWriteReplicaMap,  isFinalized);
        }
      }
    
    
    private void addReplicaToReplicasMap(Block block, ReplicaMap volumeMap,
          final RamDiskReplicaTracker lazyWriteReplicaMap,boolean isFinalized)
          throws IOException {
        ReplicaInfo newReplica = null;
        long blockId = block.getBlockId();
        long genStamp = block.getGenerationStamp();
    
        //如果是finalized文件夹， 则将所有数据块加载为FINALIZED状态
        if (isFinalized) {
          newReplica = new ReplicaBuilder(ReplicaState.FINALIZED)
              .setBlockId(blockId)
              .setLength(block.getNumBytes())
              .setGenerationStamp(genStamp)
              .setFsVolume(volume)
              .setDirectoryToUse(DatanodeUtil.idToBlockDir(finalizedDir, blockId))
              .build();
        } else {
          File file = new File(rbwDir, block.getBlockName());
          boolean loadRwr = true;
          File restartMeta = new File(file.getParent()  +
              File.pathSeparator + "." + file.getName() + ".restart");
          Scanner sc = null;
          try {
            //重启元数据文件存在， 并且当前时间在重启窗口内
            sc = new Scanner(restartMeta, "UTF-8");
            // The restart meta file exists
            if (sc.hasNextLong() && (sc.nextLong() > timer.now())) {
              // It didn't expire. Load the replica as a RBW.
              // We don't know the expected block length, so just use 0
              // and don't reserve any more space for writes.
              //将数据块加载为RBW状态
              newReplica = new ReplicaBuilder(ReplicaState.RBW)
                  .setBlockId(blockId)
                  .setLength(validateIntegrityAndSetLength(file, genStamp))
                  .setGenerationStamp(genStamp)
                  .setFsVolume(volume)
                  .setDirectoryToUse(file.getParentFile())
                  .setWriterThread(null)
                  .setBytesToReserve(0)
                  .build();
              loadRwr = false;
            }
            sc.close();
            if (!fileIoProvider.delete(volume, restartMeta)) {
              FsDatasetImpl.LOG.warn("Failed to delete restart meta file: " +
                  restartMeta.getPath());
            }
          } catch (FileNotFoundException fnfe) {
            // nothing to do hereFile dir =
          } finally {
            if (sc != null) {
              sc.close();
            }
          }
          //重启元数据文件不存在， 将数据块加载为RWR状态
          // Restart meta doesn't exist or expired.
          if (loadRwr) {
            ReplicaBuilder builder = new ReplicaBuilder(ReplicaState.RWR)
                .setBlockId(blockId)
                .setLength(validateIntegrityAndSetLength(file, genStamp))
                .setGenerationStamp(genStamp)
                .setFsVolume(volume)
                .setDirectoryToUse(file.getParentFile());
            newReplica = builder.build();
          }
        }
    
        ReplicaInfo tmpReplicaInfo = volumeMap.addAndGet(bpid, newReplica);
    
        //将数据块信息放入volumeMap中
        ReplicaInfo oldReplica = (tmpReplicaInfo == newReplica) ? null
            : tmpReplicaInfo;
        if (oldReplica != null) {
          // We have multiple replicas of the same block so decide which one
          // to keep.
          newReplica = resolveDuplicateReplicas(newReplica, oldReplica, volumeMap);
        }
    
        // If we are retaining a replica on transient storage make sure
        // it is in the lazyWriteReplicaMap so it can be persisted
        // eventually.
        if (newReplica.getVolume().isTransientStorage()) {
          lazyWriteReplicaMap.addReplica(bpid, blockId,
              (FsVolumeImpl) newReplica.getVolume(), 0);
        } else {
          lazyWriteReplicaMap.discardReplica(bpid, blockId, false);
        }
        if (oldReplica == null) {
          incrNumBlocks();
        }
      }

**ecoverTempUnlinkedBlock()**  
addToReplicasMap()方法调用了recoverTempUnlinkedBlock()恢复临时文件（\*.unlinked文件） 。 我们首先看一下临时文件的概念。 Datanode在升级过程中或者回滚到上一版本前， previous目录和current目录会使用硬链接保留相同数据块的引用。 如果这时客户端对  
任意数据块进行追加写操作， 将会同时影响两个引用， 也就是影响上一版本数据的快照。这时需要去除current目录对该数据块的硬链接， 保证后续对该数据块的修改不会影响previous目录中保存的上一版本数据块的快照。  
Datanode目前的逻辑是， 将需要更改的数据块先复制一份， 创建一个临时文件并命名为“$\{blockName\}.unlinked”， 这样临时文件将不会存在任何硬链接， 且内容与源文件完全相同。 当临时文件创建成功后， 再把临时文件改名为源文件。 这时current和previous中将  
存在两个不同的数据块文件， 但内容却是完全相同的。

recoverTempUnlinkedBlock()方法就是用来对临时文件（\*.unlinked文件） 进行恢复操作的。 恢复操作的逻辑是首先判断源文件是否存在， 如果不存在则将临时文件改名为源文件； 如果源文件存在， 则将临时文件直接删除， 非常简单。

/**
       * Recover an unlinked tmp file on datanode restart. If the original block
       * does not exist, then the tmp file is renamed to be the
       * original file name and the original name is returned; otherwise the tmp
       * file is deleted and null is returned.
       */
      File recoverTempUnlinkedBlock(File unlinkedTmp) throws IOException {
        File blockFile = FsDatasetUtil.getOrigFile(unlinkedTmp);
        if (blockFile.exists()) {
          // If the original block file still exists, then no recovery is needed.
          if (!fileIoProvider.delete(volume, unlinkedTmp)) {
            //源文件存在， 则删除临时文件， 如果删除失败则抛出异常
            throw new IOException("Unable to cleanup unlinked tmp file " +
                unlinkedTmp);
          }
          return null;
        } else {
          //源文件不存在， 则将临时文件改名为源文件
          fileIoProvider.rename(volume, unlinkedTmp, blockFile);
          return blockFile;
        }
      }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/ff71f419172042dd80fbcd6950d38544.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/f30a299ad1604607b2e022f0a5c8e56d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201110093541123.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201110135828360.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center