HDFS源码分析之UnderReplicatedBlocks(二)

简介:         UnderReplicatedBlocks还提供了一个数据块迭代器BlockIterator,用于遍历其中的数据块。它是UnderReplicatedBlocks的内部类,有三个成员变量,如下: // 当前迭代级别 private int level; ...

        UnderReplicatedBlocks还提供了一个数据块迭代器BlockIterator,用于遍历其中的数据块。它是UnderReplicatedBlocks的内部类,有三个成员变量,如下:

	// 当前迭代级别
    private int level;
    
    // 标志位:是否为特定复制优先级的迭代器
    private boolean isIteratorForLevel = false;
    
    // 数据块Block迭代器Iterator列表,存储各级别数据块迭代器
    private final List<Iterator<Block>> iterators = new ArrayList<Iterator<Block>>();
        其中,level代表了迭代器当前处于的迭代级别,表示正在哪个块复制级别迭代数据块;isIteratorForLevel是一个标志位,是否为特定复制优先级的迭代器的标志位,也就意味着只在特定级别进行迭代;而iterators则是一个数据块Block迭代器Iterator列表,由前往后、由高到低的存储各级别数据块迭代器。
        BlockIterator提供了两个构造函数,一个是无参构造函数:生成所有级别的数据块迭代器,另外一个是有参构造函数:生成指定级别的数据块迭代器,代码分别如下:

    /**
     * Construct an iterator over all queues.
     * 无参构造函数:生成所有级别的数据块迭代器
     */
    private BlockIterator() {
      // 当前迭代级别level设置为0
      level=0;
      // iterators中添加全部级别的数据块迭代器
      for(int i=0; i<LEVEL; i++) {
        iterators.add(priorityQueues.get(i).iterator());
      }
    }

    /**
     * Constrict an iterator for a single queue level
     * 有参构造函数:生成指定级别的数据块迭代器
     * @param l the priority level to iterate over
     */
    private BlockIterator(int l) {
      // 当前迭代级别level设置为指定级别
      level = l;
      // 标志位:是否为特定复制优先级的迭代器设置为true
      isIteratorForLevel = true;
      // iterators中添加指定级别的数据块迭代器
      iterators.add(priorityQueues.get(level).iterator());
    }
        注释很清晰,读者可自行阅读。另外,数据块是根据复制级别由高到低的顺序迭代的,当某一级别数据块迭代完毕,那么我们需要更新当前迭代级别,此时update()方法就完成这一个工作,代码如下:

    // 如果需要,更新当前迭代级别(由高往低迭代)
    private void update() {
    	
      // 标志位:是否为特定复制优先级的迭代器,为true的话,直接返回
      if (isIteratorForLevel) {
        return;
      }
      
      // 当前迭代级别小于LEVEL-1(也就是还没到最低级别),并且当前迭代级别已遍历所有数据块
      while(level< LEVEL-1 && !iterators.get(level).hasNext()) {
    	// 当前迭代级别level加1
        level++;
      }
    }
        它会先判断标志位isIteratorForLevel:是否为特定复制优先级的迭代器,为true的话,直接返回;否则当当前迭代级别小于LEVEL-1(也就是还没到最低级别),并且当前迭代级别已遍历所有数据块,当前迭代级别level加1。

        既然是一个迭代器,那么我们看下它最重要的两个方法,hasNext()和next(),分别如下:

        hasNext():判断是否还存在未迭代元素

    // 判断是否还有下一个元素
    @Override
    public boolean hasNext() {
    	
      // 标志位:是否为特定复制优先级的迭代器,为true的话,直接返回iterators列表中第一个迭代器的判断下一个元素的结果
      if (isIteratorForLevel) {
        return iterators.get(0).hasNext();
      }
      
      // 如果需要,更新当前迭代级别(由高往低迭代)
      update();
      
      // 取当前级别的迭代器的判断是否存在下一个元素的结果
      return iterators.get(level).hasNext();
    }
        hasNext()方法,也是先判断标志位isIteratorForLevel:是否为特定复制优先级的迭代器,为true的话,直接返回iterators列表中第一个迭代器的判断下一个元素的结果,否则,如果需要,调用update()方法更新当前迭代级别(由高往低迭代),取当前级别的迭代器的判断是否存在下一个元素的结果。

        next()方法:跌倒下一个元素

    // 取下一个元素
    @Override
    public Block next() {
    	
      // 标志位:是否为特定复制优先级的迭代器,为true的话,直接返回iterators列表中第一个迭代器的下一个元素
      if (isIteratorForLevel) {
        return iterators.get(0).next();
      }
      
      // 如果需要,更新当前迭代级别(由高往低迭代)
      update();
      
      // 取当前级别的迭代器的下一个元素
      return iterators.get(level).next();
    }
        处理逻辑与hasNext()方法一致,不再赘述。

        BlockIterator还提供了从迭代器中移除元素remove()方法及获取当前迭代级别的getPriority()方法,代码分别如下:

    // 移除
    @Override
    public void remove() {
    	
      // 标志位:是否为特定复制优先级的迭代器,为true的话,直接返回iterators列表中第一个迭代器的移除结果
      if (isIteratorForLevel) {
        iterators.get(0).remove();
      } else {
    	// 取当前级别的迭代器的移除的结果
        iterators.get(level).remove();
      }
    }

    // 获取当前迭代级别
    int getPriority() {
      return level;
    }


        UnderReplicatedBlocks还提供了按照优先级由高到低的顺序,获取指定数目的待复制数据块的chooseUnderReplicatedBlocks()方法,代码如下:

  /**
   * Get a list of block lists to be replicated. The index of block lists
   * represents its replication priority. Replication index will be tracked for
   * each priority list separately in priorityToReplIdx map. Iterates through
   * all priority lists and find the elements after replication index. Once the
   * last priority lists reaches to end, all replication indexes will be set to
   * 0 and start from 1st priority list to fulfill the blockToProces count.
   * 
   * @param blocksToProcess - number of blocks to fetch from underReplicated blocks.
   * @return Return a list of block lists to be replicated. The block list index
   *         represents its replication priority.
   */
  public synchronized List<List<Block>> chooseUnderReplicatedBlocks(
      int blocksToProcess) {
    
	// initialize data structure for the return value
	// 初始化做为返回值的数据结构,LEVEL大小的一个块列表的列表
    List<List<Block>> blocksToReplicate = new ArrayList<List<Block>>(LEVEL);
   
    // 每种优先级都添加一个空的数据块ArrayList列表
    for (int i = 0; i < LEVEL; i++) {
      blocksToReplicate.add(new ArrayList<Block>());
    }

    // 如果不存在需要复制的数据块,直接返回空的列表blocksToReplicate
    if (size() == 0) { // There are no blocks to collect.
      return blocksToReplicate;
    }
    
    int blockCount = 0;
    
    // 按照优先级从高到低处理
    for (int priority = 0; priority < LEVEL; priority++) { 
      
      // Go through all blocks that need replications with current priority.
      // 构造指定级别priority的数据块迭代器BlockIterator实例neededReplicationsIterator
      BlockIterator neededReplicationsIterator = iterator(priority);
      
      // 根据指定级别priority从priorityToReplIdx中取之前已经处理到的位置索引replIndex
      Integer replIndex = priorityToReplIdx.get(priority);
      
      // skip to the first unprocessed block, which is at replIndex
      // 利用replIndex,数据块迭代器跳过之前已处理的数据块,指向下一个该处理的正确位置
      for (int i = 0; i < replIndex && neededReplicationsIterator.hasNext(); i++) {
        neededReplicationsIterator.next();
      }

      // blocksToProcess的值不能超过所有待复制数据块总数
      blocksToProcess = Math.min(blocksToProcess, size());
      
      // 如果已获取到足够的数据块,即blockCount等于blocksToProcess,直接跳出循环
      if (blockCount == blocksToProcess) {
        break;  // break if already expected blocks are obtained
      }
      
      // Loop through all remaining blocks in the list.
      // 通过迭代器neededReplicationsIterator迭代数据块,添加入返回集合blocksToReplicate中,
      // 并累加索引位置replIndex
      
      // 判断条件为当前已取数据块blockCount还未达到要求的blocksToProcess,
      // 同时数据块迭代器neededReplicationsIterator还有下一个元素
      while (blockCount < blocksToProcess
          && neededReplicationsIterator.hasNext()) {
        
    	// 通过数据块迭代器neededReplicationsIterator取下一个数据块
    	Block block = neededReplicationsIterator.next();
    	
    	// 将数据块添加到优先级priority对应的列表
        blocksToReplicate.get(priority).add(block);
        
        // 累加索引位置replIndex
        replIndex++;
        
        // 累加已获取数据块数目blockCount
        blockCount++;
      }
      
      // 如果迭代器中已没有元素,且已处理到最高级别,重置位置索引priorityToReplIdx为0,跳出循环
      if (!neededReplicationsIterator.hasNext()
          && neededReplicationsIterator.getPriority() == LEVEL - 1) {
        // reset all priorities replication index to 0 because there is no
        // recently added blocks in any list.
        for (int i = 0; i < LEVEL; i++) {
          priorityToReplIdx.put(i, 0);
        }
        break;
      }
      
      // 记录索引位置replIndex
      priorityToReplIdx.put(priority, replIndex); 
    }
    
    // 返回获取到的数据块列表
    return blocksToReplicate;
  }

        其处理逻辑大体如下:

        1、初始化做为返回值的数据结构,LEVEL大小的一个块列表的列表blocksToReplicate;

        2、每种优先级都添加一个空的数据块ArrayList列表;

        3、如果不存在需要复制的数据块,直接返回空的列表blocksToReplicate;

        4、按照优先级从高到低处理:

              4.1、构造指定级别priority的数据块迭代器BlockIterator实例neededReplicationsIterator;

              4.2、根据指定级别priority从priorityToReplIdx中取之前已经处理到的位置索引replIndex;

              4.3、利用replIndex,数据块迭代器跳过之前已处理的数据块,指向下一个该处理的正确位置;

              4.4、blocksToProcess的值不能超过所有待复制数据块总数;

              4.5、如果已获取到足够的数据块,即blockCount等于blocksToProcess,直接跳出循环;

              4.6、通过迭代器neededReplicationsIterator迭代数据块,添加入返回集合blocksToReplicate中,并累加索引位置replIndex,判断条件为当前已取数据块blockCount还未达到要求的blocksToProcess,同时数据块迭代器neededReplicationsIterator还有下一个元素:

                       4.6.1、通过数据块迭代器neededReplicationsIterator取下一个数据块;

                       4.6.2、将数据块添加到优先级priority对应的列表;

                       4.6.3、累加索引位置replIndex;

                       4.6.4、累加已获取数据块数目blockCount;

              4.7、如果迭代器中已没有元素,且已处理到最高级别,重置位置索引priorityToReplIdx为0,跳出循环;

              4.8、priorityToReplIdx中记录优先级priority对应索引位置replIndex;

        5、返回获取到的数据块列表。

        未完待续,敬请期待《HDFS源码分析之UnderReplicatedBlocks(二)》





相关文章
|
存储 缓存 分布式计算
|
存储 机器学习/深度学习 分布式计算
|
存储 调度 机器学习/深度学习
HDFS源码分析心跳汇报之数据块汇报
        在《HDFS源码分析心跳汇报之数据块增量汇报》一文中,我们详细介绍了数据块增量汇报的内容,了解到它是时间间隔更长的正常数据块汇报周期内一个smaller的数据块汇报,它负责将DataNode上数据块的变化情况及时汇报给NameNode。
1113 0
HDFS源码分析心跳汇报之DataNode注册
        HDFS源码分析心跳汇报之DataNode注册,近期推出!
888 0
|
机器学习/深度学习
HDFS源码分析心跳汇报之整体结构
        我们知道,HDFS全称是Hadoop Distribute FileSystem,即Hadoop分布式文件系统。既然它是一个分布式文件系统,那么肯定存在很多物理节点,而这其中,就会有主从节点之分。
1240 0
|
机器学习/深度学习
HDFS源码分析心跳汇报之数据结构初始化
        在《HDFS源码分析心跳汇报之整体结构》一文中,我们详细了解了HDFS中关于心跳的整体结构,知道了BlockPoolManager、BPOfferService和BPServiceActor三者之间的关系。
1125 0
|
存储 缓存
HDFS源码分析DataXceiver之读数据块
         在《HDFS源码分析DataXceiver之整体流程》一文中我们知道,无论来自客户端还是其他数据节点的请求达到DataNode时,DataNode上的后台线程DataXceiverServer均为每个请求创建一个单独的后台工作线程来处理,这个工作线程就是DataXceiver。
1199 0
|
存储 分布式计算 Hadoop
HDFS源码分析DataXceiver之整体流程
        在《HDFS源码分析之DataXceiverServer》一文中,我们了解到在DataNode中,有一个后台工作的线程DataXceiverServer。它被用于接收来自客户端或其他数据节点的数据读写请求,为每个数据读写请求创建一个单独的线程去处理。
1545 0