第七章：小朱笔记hadoop之源码分析-hdfs分析 第四节：namenode-ReplicationMonitor

第四节：namenode分析

4.3 namenode 副本监控分析ReplicationMonitor

ReplicationMonitor主要有两个作用：

（1）负责为副本不足的数据块选择source 数据节点，选择冗余的target节点，等待DN节点下次心跳将这些工作带回给相应的DN执行块冗余操作。
（2）将各个数据节点上无效的数据块副本加入无效集合，等待下次心跳将这些工作带回给相应的DataNode执行删除无效块操作。
默认每3s执行一次，可以通过修改dfs.replication.interval来调整执行间隔
（1）computeDatanodeWork
    计算datanode需要处理的replication数量，主要包括当前超时挂起的replication，需要进行复制的replication，计 划处理的replication,损坏的replication。将这些数据记录下来，在下次heartbeat时候通知给datanode处理。
（2）processPendingReplications
    处理超时挂起的replication，将超时的replication加入到需要进行replication操作的队列中
    重要的控制参数：

//限制每一次处理无效Blocks的数据节点数量  
static final int INVALIDATE_WORK_PCT_PER_ITERATION = 32;   
//限制每一次处理冗余Blocks的数据节点数量  
static final float REPLICATION_WORK_MULTIPLIER_PER_ITERATION = 2;

（2）PendingReplicationBlocks pendingReplications

     描述当前尚未完成块副本复制的块的列表。因为HDFS支持块的流水线复制，当文件系统客户端开始向第一个Datanode复制数据块的时候，会在第一个 Datanode上启动复制进程，将该结点上接收到的（部分），数据块开始向第二个Datanode上传输复制，而第二个Datanode结点又向第三个 Datanode结点进行流水线复制，……，直到满足副本因子要求。所以，在执行流水线复制的过程中，因为这种可并行的复制方式使得某些块副本的复制完成 时间呈现阶梯状，也就是使用一个数据结构来保存这些尚未复制完成的块副本pendingReplications保存了所有正在进行复制的数据块，使用 Map是需要一些附加的信息PendingBlockInfo。这些信息包括时间戳， 用于检测是否已经超时，和现在进行复制的数目numReplicasInProgress。timedOutItems是超时的复制项，超时的复制项在 FSNamesystem的processPendingReplications方法中被删除，并从新复制。timerThread是用于检测复制超时 的线程的句柄，对应的线程是PendingReplicationMonitor的一个实例，它的run方法每隔一段会检查是否有超时的复制项，如果有， 将该数据块加到timedOutItems中。Timeout是run方法的检查间隔，defaultRecheckInterval是缺省值。在 ReplicationMonitor中，又会将timedOutItems中的数据块重新加入neededReplications；

我们分两个方面一一分析：
（1）块冗余处理机制
          块冗余处理机制分为四个步骤：选择需要冗余的数据块、为数据块选择源节点、为数据块选择目标节点、将块和目标节点加入待冗余的队列。下面一一分析：

  第一步：获取一定量需要冗余的数据块
    当DN节点报告块的时候，NN会判断块的冗余是否足够，如果不足可能有几种情况：副本可能1份或者2份，如果副本仅一份，则需要冗余的优先级就相对更高， 则会将该块放入优先级更高的队列，副本已经有2份的块则会放入优先级更低点的队列，编号越低，一共分为3个优先级。
UnderReplicatedBlocks 主要的成员是一个优先队列，List<TreeSet<Block>> priorityQueues，保存副本数没有达到期望值的block。根据当前副本数低于期望值的程度决定优先级，差的越远，优先级越大（0~2,0最 大）。每个优先级别对应一个TreeSet，getPriority获得优先级后决定放入哪个treeset中。
      对于冗余工作，根据DN节点数量不同，每次处理的数据块也会不同，比如当前集群有10个DN 节点，则默认情况下，ReplicationMonitor一个周期只会从neededReplications中取出20个数据块进行处理，之所以 HDFS要这样设计，主要是考虑到如果一次冗余的数据块太多，势必会对DN节点的网络造成比较大的影响，因此根据不同的DN集群规模决定一次冗余的数据块 数量。在具体冗余的过程中，首先按照优先级从优先级队列中拿一部分数据（如前面提及的20）块到replicateBlocks中，并且每次记录下读取的 优先级队列的偏移量，下次从该偏移指针处真正读取块信息。

      以上工作均在 chooseUnderReplicatedBlocks中分四步完成：

synchronized List<List<Block>> chooseUnderReplicatedBlocks(int blocksToProcess) {  
        // initialize data structure for the return value  
        //1 初始化三级数组  
        List<List<Block>> blocksToReplicate =  new ArrayList<List<Block>>(UnderReplicatedBlocks.LEVEL);  
        for (int i=0; i<UnderReplicatedBlocks.LEVEL; i++) {  
          blocksToReplicate.add(new ArrayList<Block>());  
        }  
          
        synchronized(neededReplications) {  
          if (neededReplications.size() == 0) {  
            missingBlocksInCurIter = 0;  
            missingBlocksInPrevIter = 0;  
            return blocksToReplicate;  
          }  
            
          //2.跳过已经处理的数据 replIndex为游标  
          // Go through all blocks that need replications.  
          BlockIterator neededReplicationsIterator = neededReplications.iterator();  
          // skip to the first unprocessed block, which is at replIndex   
          for(int i=0; i < replIndex && neededReplicationsIterator.hasNext(); i++) {  
            neededReplicationsIterator.next();  
          }  
          // # of blocks to process equals either twice the number of live   
          // data-nodes or the number of under-replicated blocks whichever is less  
          //要取的block 和 neededReplications 数量 取较小值  
          blocksToProcess = Math.min(blocksToProcess, neededReplications.size());  
      
          //3.迭代获取需要冗余的Block，根据优先级放入不同的数据  
          for (int blkCnt = 0; blkCnt < blocksToProcess; blkCnt++, replIndex++) {  
            if( ! neededReplicationsIterator.hasNext()) {  
              // start from the beginning  
              replIndex = 0;  
              missingBlocksInPrevIter = missingBlocksInCurIter;  
              missingBlocksInCurIter = 0;  
              blocksToProcess = Math.min(blocksToProcess, neededReplications.size());  
              if(blkCnt >= blocksToProcess)  
                break;  
              neededReplicationsIterator = neededReplications.iterator();  
              assert neededReplicationsIterator.hasNext() :   
                                      "neededReplications should not be empty.";  
            }  
      
            Block block = neededReplicationsIterator.next();  
            int priority = neededReplicationsIterator.getPriority();  
            if (priority < 0 || priority >= blocksToReplicate.size()) {  
              LOG.warn("Unexpected replication priority: " + priority + " " + block);  
            } else {  
              blocksToReplicate.get(priority).add(block);  
            }  
          } // end for  
        } // end synchronized  
        return blocksToReplicate;  
     }

第二步：选择源节点

private DatanodeDescriptor chooseSourceDatanode(Block block,List<DatanodeDescriptor> containingNodes,NumberReplicas numReplicas);  
      
            依赖的数据结构  
        （a）块（Block）到其元数据的映射表，元数据信息包括块所属的inode、存储块的Datanode  
        BlocksMap blocksMap   
      
        （b）保存损坏（如：校验没通过）的数据块到对应DataNode的关系   
          CorruptReplicasMap corruptReplicas   
      
            （c）保存Datanode上有效但需要删除的数据块（StorageID -> TreeSet<Block>）   
        Map<String, Collection<Block>> excessReplicateMap

 主要逻辑：
    （a）判断在某DataNode上副本是否损坏，如果损坏，遍历下个DataNode节点；
    （b）判断在某DataNode对于的replicateBlocks中，是否已经有大于等于2个数据块了，如果是这样，遍历下一个DataNode节点，之所以这样实现，防止单DataNode流量过高；
    （c）如果某DataNode节点已经退役，遍历下一个DataNode节点；
    （d）如果某DataNode节点正在退役，将该DataNode作为source，而这也是HDFS设计的目的，这样的节点不会让出现因为读写数据块而 网络拥 塞，因此也不会让网络那么忙，当然即使是正在退役的节点，replicateBlocks中的块数量也不能超过2个（防止网络拥塞）；如果有多个节点正在 退役，会选择最后一个退役节点作为该块的source节点；
    （e）如果没有正在退役的节点，随意选择一个DN节点作为source节点；

第三步：选择目标节点       目前存储了该块副本的DN节点不会再作为target节点，至于选择target节点的数目，根据需要再冗余的块副本数量而定，比如还需要冗余2个副 本，则会选择2个target节点，选择的方式，依然采用机架感知相关算法。该算法我会专门用一节来详细讲述。

第四步：将块和目标节点加入待冗余的队列

srcNode.addBlockToBeReplicated(block, targets);
注意：DatanodeDescriptor中的BlockQueue replicateBlocks = new BlockQueue();

（2）块无效处理机制

  这种类型数据块对于集群没有作用，如果不删除，会占用磁盘空间。在几种情况下会产生无效块：
（a）执行删除文件的时候，会首先快速地删除元数据，文件所对应的数据块就变得无效；
（b）Client上传数据块到DataNode的时候，可能由于网络原因等，数据块上传一部分后失败，这些块成了无效块；
（c） 在HDFS做rebalance操作的时候，会将负载较重的DataNode上部分块文件复制到其他负载较轻的DataNode上，数据块复制成功后，负 载较轻的DataNode上的该数据块就成了无效块；处理的无效块数量也是有限制的：数量为活着的dn的32%, 即 nodesToProcess = (int)Math.ceil((double)heartbeats.size() * ReplicationMonitor.INVALIDATE_WORK_PCT_PER_ITERATION / 100); 在该流程中，也并不是无效块集合中有多少块，NameNode就一次性让DataNode把相应的无效块删除完，而是每次均删除一部分，这样的话不会造成 DataNode因为删除块而带宽，CPU资源等占用过大。
依赖数据结构： Map<String, Collection<Block>> recentInvalidateSets ，保存了每个DataNode上有效，但需要删除的数据块（StorageID >> TreeSet<Block>）。

流程分析：
（a）周期性调用FSNameSystem的computeDatanodeWork方法
（b）computeDatanodeWork调用computeInvalidateWork方法
（c）循环调用invalidateWorkForOneNode方法，直到达到预设值或全部处理完
（d）invalidateWorkForOneNode方法中，从recentInvalidateSets结构中获得第一个datanode节点，找到 对应的DatanodeDescriptro，同时将其所有对应invalidate block遍历出
（e）将遍历出的invalidate block封装调用addBlocksToBeInvalidated方法，在下一次心跳汇报时返回给datanode进行相应处理