投稿
  1. 云图网首页
  2. 技术专区
  3. 大数据

HDFS源码分析(三)—–数据块关系基本结构详解大数据

• 大数据 • 阅读 47

前言

正如我在前面的文章中曾经写过，在HDFS中存在着两大关系模块，一个是文件与block数据块的关系，简称为第一关系，但是相比于第一个关系清晰的结构关系，HDFS的第二关系就没有这么简单了，第二关系自然是与数据节点相关，就是数据块与数据节点的映射关系，里面的有些过程的确是错综复杂的，这个也很好理解嘛，本身block块就很多，而且还有副本设置，然后一旦集群规模扩大，数据节点的数量也将会变大，如何处理此时的数据块与对应数据节点的映射就必然不是简单的事情了，所以这里有一点是比较特别的，随着系统的运行，数据节点与他所包含的块列表时动态建立起来的，相应的名字节点也需要通过心跳来获取信息，并不断更新元数据信息。同样在第二关系中,有诸多巧妙的设计,望读者在下面的阅读中细细体会

相关涉及类

数据块关系中涉及的基本类其实不是很多,见如下:

1.BlocksMap–数据块映射图类,保存了数据块到数据节点列表以及所属INode文件节点的映射关系,是非常重要的一个类.

2.BlocksMap.BlockInfo–BlocksMap的内部类,保存了block的数据信息类.

3.DatanodeDescriptor–数据节点类,是对数据节点的抽象,里面包含了许多与数据节点有关的变量信息.

4.DatanodeID和DatanodeInfo–数据节点基本类,是DatanodeDescriptor的父类,在这里定义了一个节点最基本的信息.

5.GSet以及LightWeightGSet–链表集合类,保存了blockInfo信息的集合类,采用了哈希存储的方式存储block数据块信息.

6.PendingReplicationBlocks和UnderReplicatedBlocks–副本相关类,其实这并不算是基本数据块关系结构中的部分,这是副本关系中涉及的类,因为与数据块有联系,放在这儿举个例子

7.FSNamesystem–命名系统类,他是一个非常大的类,里面的代码量多大5千多行,涉及了多个模块间的交互处理,其中副本相关的部分也有在这里操作的.

OK,下面开始主要内容的讲述.

BlocksMap

映射关系图类,首先这一定是类似于map这种数据类型的,所以肯定有类似于put(key,value),get(key)这样的操作,的确是这样,但是他并没有直接沿用hashMap这样的现成的类,而是自己实现了一个轻量级的集合类,什么叫做轻量级呢,与我们平常见的又有区别呢.先看BlocksMap的主要变量定义

[java] 
view plain
copy
print
?

  1. /** 
  2.  * This class maintains the map from a block to its metadata. 
  3.  * block’s metadata currently includes INode it belongs to and 
  4.  * the datanodes that store the block. 
  5.  */  
  6. class BlocksMap {  
  7. /** 
  8.    * Internal class for block metadata. 
  9.    * blockMap内部类保存block信息元数据 
  10.    */  
  11.   static class BlockInfo extends Block implements LightWeightGSet.LinkedElement {  
  12. …  
  13. }  
  14. /** Constant [email protected] LightWeightGSet} capacity. */  
  15.   private final int capacity;  
  16.     
  17.   private GSet<Block, BlockInfo> blocks;  
  18.   
  19.   BlocksMap(int initialCapacity, float loadFactor) {  
  20.     this.capacity = computeCapacity();  
  21.     //用轻量级的GSet实现block与blockInfo的映射存储  
  22.     this.blocks = new LightWeightGSet<Block, BlockInfo>(capacity);  
  23.   }  
  24. ….  

在构造函数的上面就是这个图类,新定义的集合类型,GSet,看里面的变量关系,就很明了,从Block到BlockInfo的映射,其中注意到后者是前者的子类,所以在这里,HDFS构造了一个父类到子类的映射.于是我们要跑到GSet类下看看具体的定义声明

[java] 
view plain
copy
print
?

  1. /** 
  2.  * A [email protected] GSet} is set, 
  3.  * which supports the [email protected] #get(Object)} operation. 
  4.  * The [email protected] #get(Object)} operation uses a key to lookup an element. 
  5.  *  
  6.  * Null element is not supported. 
  7.  *  
  8.  * @param <K> The type of the keys. 
  9.  * @param <E> The type of the elements, which must be a subclass of the keys. 
  10.  * 定义了特殊集合类GSet，能够通过key来寻找目标对象，也可以移除对象，对象类同时也是key的子类 
  11.  */  
  12. public interface GSet<K, E extends K> extends Iterable<E> {  
  13.   /** 
  14.    * @return The size of this set. 
  15.    */  
  16.   int size();  
  17.   boolean contains(K key);  
  18.   E get(K key);  
  19.   E put(E element);  
  20.   E remove(K key);  
  21. }  

答案的确是这样,定义的方法也是类似于map类的方法.OK,GSet的定义明白,但是这还不够,因为在构造函数,HDFS用的是他的子类LightWeightGSet,在此类的介绍文字中说,这是轻量级的集合类,可以低开销的实现元素的查找

[java] 
view plain
copy
print
?

  1. /** 
  2.  * A low memory footprint [email protected] GSet} implementation, 
  3.  * which uses an array for storing the elements 
  4.  * and linked lists for collision resolution. 
  5.  * 
  6.  * No rehash will be performed. 
  7.  * Therefore, the internal array will never be resized. 
  8.  * 
  9.  * This class does not support null element. 
  10.  * 
  11.  * This class is not thread safe. 
  12.  * 
  13.  * @param <K> Key type for looking up the elements 
  14.  * @param <E> Element type, which must be 
  15.  *       (1) a subclass of K, and 
  16.  *       (2) implementing [email protected] LinkedElement} interface. 
  17.  * 轻量级的集合类，低内存的实现用于寻找对象类，不允许存储null类型，存储的类型必须为key的子类 
  18.  * 在里面用了哈希算法做了映射，没有进行哈希重映射，因此大小不会调整 
  19.  */  
  20. public class LightWeightGSet<K, E extends K> implements GSet<K, E> {  

看起来非常神秘哦,接下来看变量定义

[java] 
view plain
copy
print
?

  1. public class LightWeightGSet<K, E extends K> implements GSet<K, E> {  
  2.   /** 
  3.    * Elements of [email protected] LightWeightGSet}. 
  4.    */  
  5.   public static interface LinkedElement {  
  6.     /** Set the next element. */  
  7.     public void setNext(LinkedElement next);  
  8.   
  9.     /** Get the next element. */  
  10.     public LinkedElement getNext();  
  11.   }  
  12.   
  13.   public static final Log LOG = LogFactory.getLog(GSet.class);  
  14.   //最大数组长度2的30次方  
  15.   static final int MAX_ARRAY_LENGTH = 1 << 30; //prevent int overflow problem  
  16.   static final int MIN_ARRAY_LENGTH = 1;  
  17.   
  18.   /** 
  19.    * An internal array of entries, which are the rows of the hash table. 
  20.    * The size must be a power of two. 
  21.    * 存储对象数组，也就是存储哈希表，必须为2的幂次方 
  22.    */  
  23.   private final LinkedElement[] entries;  
  24.   /** A mask for computing the array index from the hash value of an element. */  
  25.   //哈希计算掩码  
  26.   private final int hash_mask;  
  27.   /** The size of the set (not the entry array). */  
  28.   private int size = 0;  
  29.   /** Modification version for fail-fast. 
  30.    * @see ConcurrentModificationException 
  31.    */  
  32.   private volatile int modification = 0;  

在最上方的LinkedElement接口中有next相关的方法,可以预见到,这是一个链表结构的关系,entries数组就是存储这个链表的,链表的最大长度非常大,达到2的30次方,不过这里还有一个比较奇怪的变量hash_mask哈希掩码,这是干嘛的呢,猜测是在做取哈希索引值的时候用的.然后看下此集合的构造函数.

[java] 
view plain
copy
print
?

  1. /** 
  2.    * @param recommended_length Recommended size of the internal array. 
  3.    */  
  4.   public LightWeightGSet(final int recommended_length) {  
  5.     //传入初始长度，但是需要经过适配判断  
  6.     final int actual = actualArrayLength(recommended_length);  
  7.     LOG.info(“recommended=” + recommended_length + “, actual=” + actual);  
  8.   
  9.     entries = new LinkedElement[actual];  
  10.     //把数组长度减1作为掩码  
  11.     hash_mask = entries.length – 1;  
  12.   }  

在构造函数中传入的数组长度,不过需要经过方法验证,要在给定的合理范围之内,在这里,掩码值被设置成了长度值小1,这个得看了后面才能知道.在这个类Map的方法中,一定要看的方法有2个,put和get方法

[java] 
view plain
copy
print
?

  1. @Override  
  2.  public E get(final K key) {  
  3.    //validate key  
  4.    if (key == null) {  
  5.      throw new NullPointerException(“key == null”);  
  6.    }  
  7.   
  8.    //find element  
  9.    final int index = getIndex(key);  
  10.    for(LinkedElement e = entries[index]; e != null; e = e.getNext()) {  
  11.      if (e.equals(key)) {  
  12.        return convert(e);  
  13.      }  
  14.    }  
  15.    //element not found  
  16.    return null;  
  17.  }  

做法是先找到第一个索引值,一般元素如果没有被remove,第一个找到的元素就是目标值,这里重点看下哈希索引值的计算,这时候如果是我做的话,一般都是哈希取余数,就是hashCode()%N,在LightWeightGSet中给出略微不同的实现

[java] 
view plain
copy
print
?

  1. //构建索引的方法，取哈希值再与掩码进行计算，这里采用的并不是哈希取余的算法  
  2.   //与掩码计算的目的就是截取掉高位多余的数字部分使索引值落在数组存储长度范围之内  
  3.   private int getIndex(final K key) {  
  4.     return key.hashCode() & hash_mask;  
  5.   }  

简单理解为就是把哈希值排成全部由0和1组成的二进制数,然后直接数组长度段内低位的数字,因为高位的都被掩码计算成0舍掉了,用于位运算的好处是避免了十进制数字与二进制数字直接的来回转换.这个方法与哈希取余数的方法具体有什么样的效果上的差别这个我到时没有仔细比较过,不过在日后可以作为一种新的哈希链表算法的选择.
put方法就是标准的链表操作方法

[java] 
view plain
copy
print
?

  1. /** 
  2.    * Remove the element corresponding to the key, 
  3.    * given key.hashCode() == index. 
  4.    * 
  5.    * @return If such element exists, return it. 
  6.    *         Otherwise, return null. 
  7.    */  
  8.   private E remove(final int index, final K key) {  
  9.     if (entries[index] == null) {  
  10.       return null;  
  11.     } else if (entries[index].equals(key)) {  
  12.       //remove the head of the linked list  
  13.       modification++;  
  14.       size–;  
  15.       final LinkedElement e = entries[index];  
  16.       entries[index] = e.getNext();  
  17.       e.setNext(null);  
  18.       return convert(e);  
  19.     } else {  
  20.       //head != null and key is not equal to head  
  21.       //search the element  
  22.       LinkedElement prev = entries[index];  
  23.       for(LinkedElement curr = prev.getNext(); curr != null; ) {  
  24.          //下面是标准的链表操作  
  25.         if (curr.equals(key)) {  
  26.           //found the element, remove it  
  27.           modification++;  
  28.           size–;  
  29.           prev.setNext(curr.getNext());  
  30.           curr.setNext(null);  
  31.           return convert(curr);  
  32.         } else {  
  33.           prev = curr;  
  34.           curr = curr.getNext();  
  35.         }  
  36.       }  
  37.       //element not found  
  38.       return null;  
  39.     }  
  40.   }  

在这里不做过多介绍.再回过头来看BlockInfo内部，有一个Object对象数组

[java] 
view plain
copy
print
?

  1. static class BlockInfo extends Block implements LightWeightGSet.LinkedElement {  
  2.     //数据块信息所属的INode文件节点  
  3.     private INodeFile          inode;  
  4.   
  5.     /** For implementing [email protected] LightWeightGSet.LinkedElement} interface */  
  6.     private LightWeightGSet.LinkedElement nextLinkedElement;  
  7.   
  8.     /** 
  9.      * This array contains triplets of references. 
  10.      * For each i-th data-node the block belongs to 
  11.      * triplets[3*i] is the reference to the DatanodeDescriptor 
  12.      * and triplets[3*i+1] and triplets[3*i+2] are references  
  13.      * to the previous and the next blocks, respectively, in the  
  14.      * list of blocks belonging to this data-node. 
  15.      * triplets对象数组保存同一数据节点上的连续的block块。triplets[3*i]保存的是当前的数据节点 
  16.      * triplets[3*i+1]和triplets[3*i+2]保存的则是一前一后的block信息 
  17.      */  
  18.     private Object[] triplets;  

在这个对象数组中存储入3个对象，第一个是数据块所在数据节点对象，第二个对象保存的是数据块的前一数据块，第三对象是保存数据块的后一数据块，形成了一个双向链表的结构，也就是说，顺着这个数据块遍历，你可以遍历完某个数据节点的所有的数据块，triplets的数组长度受block数据块的副本数限制，因为不同的副本一般位于不同的数据节点。

[java] 
view plain
copy
print
?

  1. BlockInfo(Block blk, int replication) {  
  2.       super(blk);  
  3.       //因为还需要同时保存前后的数据块信息，所以这里会乘以3  
  4.       this.triplets = new Object[3*replication];  
  5.       this.inode = null;  
  6.     }  

因为存储对象的不同，所以这里用了Object作为对象数组，而不是用具体的类名。里面的许多操作也是完全类似于链表的操作

[java] 
view plain
copy
print
?

  1. /** 
  2.      * Remove data-node from the block. 
  3.      * 移除数据节点操作，把目标数据块移除，移除位置用最后一个块的信息替代，然后移除末尾块 
  4.      */  
  5.     boolean removeNode(DatanodeDescriptor node) {  
  6.       int dnIndex = findDatanode(node);  
  7.       if(dnIndex < 0) // the node is not found  
  8.         return false;  
  9.       assert getPrevious(dnIndex) == null && getNext(dnIndex) == null :   
  10.         “Block is still in the list and must be removed first.”;  
  11.       // find the last not null node  
  12.       int lastNode = numNodes()-1;   
  13.       // replace current node triplet by the lastNode one   
  14.       //用末尾块替代当前块  
  15.       setDatanode(dnIndex, getDatanode(lastNode));  
  16.       setNext(dnIndex, getNext(lastNode));   
  17.       setPrevious(dnIndex, getPrevious(lastNode));   
  18.       //设置末尾块为空  
  19.       // set the last triplet to null  
  20.       setDatanode(lastNode, null);  
  21.       setNext(lastNode, null);   
  22.       setPrevious(lastNode, null);   
  23.       return true;  
  24.     }  

DatanodeDescriptor

下面来看另外一个关键类DatanodeDescriptor,可以说是对数据节点的抽象，首先在此类中定义了数据块到数据节点的映射类

[java] 
view plain
copy
print
?

  1. /************************************************** 
  2.  * DatanodeDescriptor tracks stats on a given DataNode, 
  3.  * such as available storage capacity, last update time, etc., 
  4.  * and maintains a set of blocks stored on the datanode.  
  5.  * 
  6.  * This data structure is a data structure that is internal 
  7.  * to the namenode. It is *not* sent over-the-wire to the Client 
  8.  * or the Datnodes. Neither is it stored persistently in the 
  9.  * fsImage. 
  10.  DatanodeDescriptor数据节点描述类跟踪描述了一个数据节点的状态信息 
  11.  **************************************************/  
  12. public class DatanodeDescriptor extends DatanodeInfo {  
  13.     
  14.   // Stores status of decommissioning.  
  15.   // If node is not decommissioning, do not use this object for anything.  
  16.   //下面这个对象只与decomission撤销工作相关  
  17.   DecommissioningStatus decommissioningStatus = new DecommissioningStatus();  
  18.   
  19.   /** Block and targets pair */  
  20.   //数据块以及目标数据节点列表映射类  
  21.   public static class BlockTargetPair {  
  22.     //目标数据块  
  23.     public final Block block;  
  24.     //该数据块的目标数据节点  
  25.     public final DatanodeDescriptor[] targets;      
  26.   
  27.     BlockTargetPair(Block block, DatanodeDescriptor[] targets) {  
  28.       this.block = block;  
  29.       this.targets = targets;  
  30.     }  
  31.   }  

然后定义了这样的队列类

[java] 
view plain
copy
print
?

  1. /** A BlockTargetPair queue. */  
  2.   //block块目标数据节点类队列  
  3.   private static class BlockQueue {  
  4.     //此类维护了BlockTargetPair列表对象  
  5.     private final Queue<BlockTargetPair> blockq = new LinkedList<BlockTargetPair>();  

然后下面又声明了一系列的与数据块相关的列表

[java] 
view plain
copy
print
?

  1. /** A queue of blocks to be replicated by this datanode */  
  2. //此数据节点上待复制的block块列表  
  3. private BlockQueue replicateBlocks = new BlockQueue();  
  4. /** A queue of blocks to be recovered by this datanode */  
  5. //此数据节点上待租约恢复的块列表  
  6. private BlockQueue recoverBlocks = new BlockQueue();  
  7. /** A set of blocks to be invalidated by this datanode */  
  8. //此数据节点上无效待删除的块列表  
  9. private Set<Block> invalidateBlocks = new TreeSet<Block>();  
  10.   
  11. /* Variables for maintaning number of blocks scheduled to be written to 
  12.  * this datanode. This count is approximate and might be slightly higger 
  13.  * in case of errors (e.g. datanode does not report if an error occurs  
  14.  * while writing the block). 
  15.  */  
  16. //写入这个数据节点的块的数目统计变量  
  17. private int currApproxBlocksScheduled = 0;  
  18. private int prevApproxBlocksScheduled = 0;  
  19. private long lastBlocksScheduledRollTime = 0;  
  20. private static final int BLOCKS_SCHEDULED_ROLL_INTERVAL = 600*1000; //10min  

下面看一个典型的添加数据块的方法，在这里添加的步骤其实应该拆分成2个步骤来看，第一个步骤是将当前数据节点添加到此block数据块管理的数据节点列表中，第二是反向关系，将数据块加入到数据节点的块列表关系中，代码如下

[java] 
view plain
copy
print
?

  1. /** 
  2.   * Add data-node to the block. 
  3.   * Add block to the head of the list of blocks belonging to the data-node. 
  4.   * 将数据节点加入到block块对应的数据节点列表中 
  5.   */  
  6.  boolean addBlock(BlockInfo b) {  
  7.     //添加新的数据节点  
  8.    if(!b.addNode(this))  
  9.      return false;  
  10.    // add to the head of the data-node list  
  11.    //将此数据块添加到数据节点管理的数据块列表中，并于当前数据块时相邻位置  
  12.    blockList = b.listInsert(blockList, this);  
  13.    return true;  
  14.  }  


数据块操作例子-副本

在这里举一个与数据块相关的样本操作过程，副本请求过程。副本相关的很多操作的协调处理都是在FSNamesystem这个大类中的，因此在这个变量中就定义了很多的相关变量

[java] 
view plain
copy
print
?

  1. public class FSNamesystem implements FSConstants, FSNamesystemMBean,  
  2.     NameNodeMXBean, MetricsSource {  
  3.   …  
  4.     
  5.   //等待复制的数据块副本数  
  6.   volatile long pendingReplicationBlocksCount = 0L;  
  7.   //已损坏的副本数数目  
  8.   volatile long corruptReplicaBlocksCount = 0L;  
  9.   //正在被复制的块副本数  
  10.   volatile long underReplicatedBlocksCount = 0L;  
  11.   //正在被调度的副本数  
  12.   volatile long scheduledReplicationBlocksCount = 0L;  
  13.   //多余数据块副本数  
  14.   volatile long excessBlocksCount = 0L;  
  15.   //等待被删除的副本数  
  16.   volatile long pendingDeletionBlocksCount = 0L;  

各种状态的副本状态计数，以及下面的相应的对象设置

[java] 
view plain
copy
print
?

  1. //  
  2.   // Keeps a Collection for every named machine containing  
  3.   // blocks that have recently been invalidated and are thought to live  
  4.   // on the machine in question.  
  5.   // Mapping: StorageID -> ArrayList<Block>  
  6.   //  
  7.   //最近无效的块列表,数据节点到块的映射  
  8.   private Map<String, Collection<Block>> recentInvalidateSets =   
  9.     new TreeMap<String, Collection<Block>>();  
  10.   
  11.   //  
  12.   // Keeps a TreeSet for every named node.  Each treeset contains  
  13.   // a list of the blocks that are “extra” at that location.  We’ll  
  14.   // eventually remove these extras.  
  15.   // Mapping: StorageID -> TreeSet<Block>  
  16.   //  
  17.   // 过剩的数据副本块图,也是数据节点到数据块的映射  
  18.   Map<String, Collection<Block>> excessReplicateMap =   
  19.     new TreeMap<String, Collection<Block>>();  
  20.   
  21.   
  22. /** 
  23.  * Store set of Blocks that need to be replicated 1 or more times. 
  24.  * Set of: Block 
  25.  */  
  26.   private UnderReplicatedBlocks neededReplications = new UnderReplicatedBlocks();  
  27.   // We also store pending replication-orders.  
  28.   //待复制的block  
  29.   private PendingReplicationBlocks pendingReplications;  

在副本状态转移相关的类，主要关注2个，一个是UnderReplicatedBlocks，另外一个是PendingReplicaionBlocks，前者是准备生产副本复制请求，而后者就是待复制请求类，在
UnderReplicatedBlocks里面，做的一个很关键的操作是确定副本复制请求的优先级，他会根据剩余副本数量以及是否是在decomission等状态的情况下，然后给出优先级，所以在变量中首先会有优先级队列设置

[java] 
view plain
copy
print
?

  1. /* Class for keeping track of under replication blocks 
  2.  * Blocks have replication priority, with priority 0 indicating the highest 
  3.  * Blocks have only one replicas has the highest 
  4.  * 此类对正在操作的副本块进行了跟踪 
  5.  * 并对此设定了不同的副本优先级队列,只有1个队列拥有最高优先级 
  6.  */  
  7. class UnderReplicatedBlocks implements Iterable<BlockInfo> {  
  8.   //定义了4种level级别的队列  
  9.   static final int LEVEL = 4;  
  10.   //损坏副本数队列在最后一个队列中  
  11.   static public final int QUEUE_WITH_CORRUPT_BLOCKS = LEVEL-1;  
  12.   //定了副本优先级队列  
  13.   private List<LightWeightLinkedSet<BlockInfo>> priorityQueues  
  14.       = new ArrayList<LightWeightLinkedSet<BlockInfo>>();  
  15.     
  16.   private final RaidMissingBlocks raidQueue;  

在这里，大致上分出了2大类队列，一个是损坏副本队列，还有一个就是正常的情况，不过是优先级不同而已。level数字越小，代表优先级越高，优先级确定的核心函数

[java] 
view plain
copy
print
?

  1. /* Return the priority of a block 
  2.    *  
  3.    * If this is a Raided block and still has 1 replica left, not assign the highest priority. 
  4.    *  
  5.    * @param block a under replication block 
  6.    * @param curReplicas current number of replicas of the block 
  7.    * @param expectedReplicas expected number of replicas of the block 
  8.    */  
  9.   private int getPriority(BlockInfo block,   
  10.                           int curReplicas,   
  11.                           int decommissionedReplicas,  
  12.                           int expectedReplicas) {  
  13.     //副本数为负数或是副本数超过预期值,都属于异常情况,归结为损坏队列中  
  14.     if (curReplicas<0 || curReplicas>=expectedReplicas) {  
  15.       return LEVEL; // no need to replicate  
  16.     } else if(curReplicas==0) {  
  17.       // If there are zero non-decommissioned replica but there are  
  18.       // some decommissioned replicas, then assign them highest priority  
  19.       //如果数据节点正位于撤销操作中,以及最高优先级  
  20.       if (decommissionedReplicas > 0) {  
  21.         return 0;  
  22.       }  
  23.       return QUEUE_WITH_CORRUPT_BLOCKS; // keep these blocks in needed replication.  
  24.     } else if(curReplicas==1) {  
  25.       //副本数只有1个的时候同样给予高优先级  
  26.       return isRaidedBlock(block) ? 1 : 0; // highest priority  
  27.     } else if(curReplicas*3<expectedReplicas) {  
  28.       return 1;  
  29.     } else {  
  30.       //其他情况给及普通优先级  
  31.       return 2;  
  32.     }  
  33.   }  

这些副本请求产生之后，就会加入到PendingReplicationBlocks的类中，在里面有相应的变量会管理这些请求信息

[java] 
view plain
copy
print
?

  1. /*************************************************** 
  2.  * PendingReplicationBlocks does the bookkeeping of all 
  3.  * blocks that are getting replicated. 
  4.  * 
  5.  * It does the following: 
  6.  * 1)  record blocks that are getting replicated at this instant. 
  7.  * 2)  a coarse grain timer to track age of replication request 
  8.  * 3)  a thread that periodically identifies replication-requests 
  9.  *     that never made it. 
  10.  * 
  11.  ***************************************************/  
  12. //此类记录了待复制请求信息  
  13. class PendingReplicationBlocks {  
  14.   //block块到副本数据块复制请求信息的映射  
  15.   private Map<Block, PendingBlockInfo> pendingReplications;  
  16.   //记录超时复制请求列表  
  17.   private ArrayList<Block> timedOutItems;  
  18.   Daemon timerThread = null;  
  19.   private volatile boolean fsRunning = true;  

就是上面的PendingBlockInfo，声明了请求产生的时间和当前此数据块的请求数

[java] 
view plain
copy
print
?

  1. /** 
  2.  * An object that contains information about a block that  
  3.  * is being replicated. It records the timestamp when the  
  4.  * system started replicating the most recent copy of this 
  5.  * block. It also records the number of replication 
  6.  * requests that are in progress. 
  7.  * 内部类包含了复制请求信息，记录复制请求的开始时间，以及当前对此块的副本请求数 
  8.  */  
  9. static class PendingBlockInfo {  
  10.     //请求产生时间  
  11.   private long timeStamp;  
  12.   //当前的进行复制请求数  
  13.   private int numReplicasInProgress;  

产生时间用于进行超时检测，请求数会与预期副本数进行对比，在这个类中会对永远不结束的复制请求进行超时检测，默认时间5~10分钟

[java] 
view plain
copy
print
?

  1. //  
  2.   // It might take anywhere between 5 to 10 minutes before  
  3.   // a request is timed out.  
  4.   //  
  5.   //默认复制请求超时检测5到10分钟  
  6.   private long timeout = 5 * 60 * 1000;  
  7.   private long defaultRecheckInterval = 5 * 60 * 1000;  

下面是监控方法

[java] 
view plain
copy
print
?

  1. /* 
  2.    * A periodic thread that scans for blocks that never finished 
  3.    * their replication request. 
  4.    * 一个周期性的线程监控不会结束的副本请求 
  5.    */  
  6.   class PendingReplicationMonitor implements Runnable {  
  7.     public void run() {  
  8.       while (fsRunning) {  
  9.         long period = Math.min(defaultRecheckInterval, timeout);  
  10.         try {  
  11.           //调用下面的函数进行检查  
  12.           pendingReplicationCheck();  
  13.           Thread.sleep(period);  
  14.         } catch (InterruptedException ie) {  
  15.           FSNamesystem.LOG.debug(  
  16.                 “PendingReplicationMonitor thread received exception. “ + ie);  
  17.         }  
  18.       }  
  19.     }  
  20.   
  21.     /** 
  22.      * Iterate through all items and detect timed-out items 
  23.      */  
  24.     void pendingReplicationCheck() {  
  25.       synchronized (pendingReplications) {  
  26.         Iterator iter = pendingReplications.entrySet().iterator();  
  27.         long now = FSNamesystem.now();  
  28.         FSNamesystem.LOG.debug(“PendingReplicationMonitor checking Q”);  
  29.         while (iter.hasNext()) {  
  30.           Map.Entry entry = (Map.Entry) iter.next();  
  31.           PendingBlockInfo pendingBlock = (PendingBlockInfo) entry.getValue();  
  32.           //取出请求开始时间比较时间是否超时  
  33.           if (now > pendingBlock.getTimeStamp() + timeout) {  
  34.             Block block = (Block) entry.getKey();  
  35.             synchronized (timedOutItems) {  
  36.               //加入超时队列中  
  37.               timedOutItems.add(block);  
  38.             }  
  39.             FSNamesystem.LOG.warn(  
  40.                 “PendingReplicationMonitor timed out block “ + block);  
  41.             iter.remove();  
  42.           }  
  43.         }  
  44.       }  
  45.     }  
  46.   }  

超时的请求会被加入到timeOutItems，这些请求一般在最好又会被插入到UnderReplicatedBlocks中。OK,简单的阐述了一下副本数据块关系流程分析。

全部代码的分析请点击链接https://github.com/linyiqun/hadoop-hdfs,后续将会继续更新HDFS其他方面的代码分析。

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/tech/bigdata/9185.html

赞 (0)
0 0
上一篇 2021年7月19日 09:17
下一篇 2021年7月19日 09:17

相关推荐

发表回复

登录后才能评论