是将一份数据保存在多个节点上。但Cache机制却要简单许多，对于cache的数据，可以随时删除丢弃，并命中cache的后果仅仅是需要访问数据源读取数据；然而副本机制却不一样，副本是不能随意丢弃的，每失去一个副本，服务质量都在下降，一旦副本数下降到一定程度，则往往服务将不再可用。
2.3.2 lease机制的分析
上节中介绍了一个lease的实例，本节进一步分析lease机制的本质。
首先给出本文对lease的定义：Lease是由颁发者授予的在某一有效期内的承诺。颁发者一旦发出lease，则无论接受方是否收到，也无论后续接收方处于何种状态，只要lease不过期，颁发者一定严守承诺；另一方面，接收方在lease的有效期内可以使用颁发者的承诺，但一旦lease过期，接收方一定不能继续使用颁发者的承诺。
由于lease是一种承诺，具体的承诺内容可以非常宽泛，可以是上节的例子中数据的正确性；也可以是某种权限，例如当需要做并发控制时，同一时刻只给某一个节点颁发lease，只有持有lease的节点才可以修改数据；也可以是某种身份，例如在primary-secondary(2.2.2 )架构中，给节点颁发lease，只有持有lease的节点才具有primary身份。Lease的承诺的内涵还可以非常宽泛，这里不再一一列举。
Lease机制具有很高的容错能力。首先，通过引入有效期，Lease机制能否非常好的容错网络异常。Lease颁发过程只依赖于网络可以单向通信，即使接收方无法向颁发者发送消息，也不影响lease的颁发。由于lease的有效期是一个确定的时间点，lease的语义与发送lease的具体时间无关，所以同一个lease可以被颁发者不断重复向接受方发送。即使颁发者偶尔发送lease失败，颁发者也可以简单的通过重发的办法解决。一旦lease被接收方成功接受，后续lease机制不再依赖于网络通信，即使网络完全中断lease机制也不受影响。再者，Lease机制能较好的容错节点宕机。如果颁发者宕机，则宕机的颁发者通常无法改变之前的承诺，不会影响lease的正确性。在颁发者机恢复后，如果颁发者恢复出了之前的lease信息，颁发者可以继续遵守lease的承诺。如果颁发者无法恢复lease信息，则只需等待一个最大的lease超时时间就可以使得所有的lease都失效，从而不破坏lease机制。例如上节中的cache系统的例子中，一旦服务器宕机，肯定不会修改元数据，重新恢复后，只需等待一个最大的lease超时时间，所有节点上的缓存信息都将被清空。对于接受方宕机的情况，颁发者不需要做更多的容错处理，只需等待lease过期失效，就可以收回承诺，实践中也就是收回之前赋予的权限、身份等。最后，lease机制不依赖于存储。颁发者可以持久化颁发过的lease信息，从而在宕机恢复后可以使得在有效期的lease继续有效。但这对于lease机制只是一个优化，如之前的分析，即使颁发者没有持久化lease信息，也可以通过等待一个最大的lease时间的方式使得之前所有颁发的lease失效，从而保证机制继续有效。
Lease机制依赖于有效期，这就要求颁发者和接收者的时钟是同步的。一方面，如果颁发者的时钟比接收者的时钟慢，则当接收者认为lease已经过期的时候，颁发者依旧认为lease有效。接收者可以用在lease到期前申请新的lease的方式解决这个问题。另一方面，如果颁发者的时钟比接收者的时钟快，则当颁发者认为lease已经过期的时候，接收者依旧认为lease有效，颁发者可能将lease颁发给其他节点，造成承诺失效，影响系统的正确性。对于这种时钟不同步，实践中的通常做法是将颁发者的有效期设置得比接收者的略大，只需大过时钟误差就可以避免对lease的有效性的影响。
2.3.3 基于lease机制确定节点状态
在分布式系统中确定一个节点是否处于正常工作状态是一个困难的问题。由于可能存在网络分化，节点的状态是无法通过网络通信来确定的。下面举一个较为具体的例子来讨论这个问题。
例2.3.1：在一个primary-secondary架构的系统中，有三个节点A、B、C互为副本，其中有一个节点为primary，且同一时刻只能有一个primary节点。另有一个节点Q负责判断节点A、B、C的状态，一旦Q发现primary异常，节点Q将选择另一个节点作为primary。假设最开始时节点A为primary，B、C为secondary。节点Q需要判断节点A、B、C的状态是否正常。 首先需要说明的是基于“心跳”(Heartbeat)的方法无法很好的解决这个问题。节点A、B、C可以周期性的向Q发送心跳信息，如果节点Q超过一段时间收不到某个节点的心跳则认为这个节点异常。这种方法的问题是假如节点Q收不到节点A的心跳，除了节点A本身的异常外，也有可能是因为节点Q与节点A之间的网络中断导致的。在工程实践中，更大的可能性不是网络中断，而是节点Q与节点A之间的网络拥塞造成的所谓“瞬断”，“瞬断”往往很快可以恢复。另一种原因甚至是节点Q的机器异常，以至于处理节点A的心跳被延迟了，以至于节点Q认为节点A没有发送心跳。假设节点A本身工作正常，但Q与节点A之间的网络暂时中断，节点A与节点B、C之间的网络正常。此时节点Q认为节点A异常，重新选择节点B作为新的primary，并通知节点A、B、C新的primary是节点B。由于节点Q的通知消息到达节点A、B、C的顺序无法确定，假如先到达B，则在这一时刻，系统中同时存在两个工作中的primary，一个是A、另一个是B。假如此时A、B都接收外部请求并与C同步数据，会产生严重的数据错误。上述即所谓“双主”问题，虽然看似这种问题出现的概率非常低，但在工程实践中，笔者不止一次见到过这样的情况发生。
上述问题的出现的原因在于虽然节点Q认为节点A异常，但节点A自己不认为自己异常，依旧作为primary工作。其问题的本质是由于网络分化造成的系统对于“节点状态”认知的不一致。
上面的例子中的分布式协议依赖于对节点状态认知的全局一致性，即一旦节点Q认为某个节点A异常，则节点A也必须认为自己异常，从而节点A停止作为primary，避免“双主”问题的出现。解决这种问题有两种思路，第一、设计的分布式协议可以容忍“双主”错误，即不依赖于对节点状态的全局一致性认识，或者全局一致性状态是全体协商后的结果；第二、利用lease机制。对于第一
种思路即放弃使用中心化的设计，而改用去中心化设计，超过本节的讨论范畴。下面着重讨论利用lease机制确定节点状态。
由中心节点向其他节点发送lease，若某个节点持有有效的lease，则认为该节点正常可以提供服务。用于例2.3.1中，节点A、B、C依然周期性的发送heart beat报告自身状态，节点Q收到heart beat后发送一个lease，表示节点Q确认了节点A、B、C的状态，并允许节点在lease有效期内正常工作。节点Q可以给primary节点一个特殊的lease，表示节点可以作为primary工作。一旦节点Q希望切换新的primary，则只需等前一个primary的lease过期，则就可以安全的颁发新的lease给新的primary节点，而不会出现“双主”问题。
在实际系统中，若用一个中心节点发送lease也有很大的风险，一旦该中心节点宕机或网络异常，则所有的节点没有lease，从而造成系统高度不可用。为此，实际系统总是使用多个中心节点互为副本，成为一个小的集群，该小集群具有高可用性，对外提供颁发lease的功能。chubby和zookeeper都是基于这样的设计。
2.3.4 lease的有效期时间选择
Lease的有效期虽然是一个确定的时间点，当颁发者在发布lease时通常都是将当前时间加上一个固定的时长从而计算出lease的有效期。如何选择Lease的时长在工程实践中是一个值得讨论的问题。如果lease的时长太短，例如1s，一旦出现网络抖动lease很容易丢失，从而造成节点失去lease，使得依赖lease的服务停止；如果lease的时长太大，例如1分钟，则一旦接受者异常，颁发者需要过长的时间收回lease承诺。例如，使用lease确定节点状态时，若lease时间过短，有可能造成网络瞬断时节点收不到lease从而引起服务不稳定，若lease时间过长，则一旦某节点宕机异常，需要较大的时间等待lease过期才能发现节点异常。工程中，常选择的lease时长是10秒级别，这是一个经过验证的经验值，实践中可以作为参考并综合选择合适的时长。
2.3.5 工程投影
本节介绍几个典型分布式系统中的运用Lease机制的情况。
2.3.5.1 GFS中的Lease
GFS中使用Lease确定Chuck的Primary副本。Lease由Master节点颁发给primary副本，持有Lease的副本成为primary副本。Primary副本控制该chuck的数据更新流量，确定并发更新操作在chuck上的执行顺序。GFS中的Lease信息由Master在响应各个节点的HeartBeat时附带传递（piggyback）。对于每一个chuck，其上的并发更新操作的顺序在各个副本上是一致的，首先master选择primary的顺序，即颁发Lease的顺序，在每一任的primary任期内，每个primary决定并发更新的顺序，从而更新操作的顺序最终全局一致。当GFS的master失去某个节点的HeartBeat时，只
看过本文章的还看过。。。
2012 年 5 月 1 1 概念 1 模型一些经典的分布式系统的资料对分布式系统的全貌做了比较详细的介绍[1][2]。为了控制规模, 在开始讨论分布式系统的协议、原理.........
分布式系统原理和课后习题答案 隐藏>> 第一章 绪论 1、中间件在分......
《分布式系统原理与范型 (第二版)》复习资料_工学_高等教育_教育专区。《分布式系统原理与范型 (第二版)》复习资料 分布式复习资料 第1章分布式系统是若干独立.........
浅析分布式文件系统原理及改进_姚毓才_互联网_it计算机_专业资料。工程科技 《 》2008 年第 6 期 浅析分布式文件系统原理及改进姚毓才 1,2 ( 安徽大学 , .........
《分布式系统原理与范型(第2版) 》课后习题答案_工学_高等教育_教育专区。分布......
分布式系统原理与范型_理学_高等教育_教育专区。分布式系统原理与范型 line ......
分布式系统原理与范例课件introduction of distributed ......
大型分布式系统理论体系_计算机软件及应用_it计算机_专业资料。大型分布式系统理论...case study: 基于 lease 的分布式 cache 基本原理:中心服务器在向各节点发送.........
《分布式系统原理与范型(第2版) 》课后习题答案_工学_高等教育_教育专区。分布......
分布式系统原理与范型课后习题答案_研究生入学考试_高等教育_教育专区。第一章 绪......
分布式系统原理与范型_理学_高等教育_教育专区。分布式系统原理与范型 line ......
《分布式系统原理与范型 (第二版)》复习资料_工学_高等教育_教育专区。《分布式系统原理与范型 (第二版)》复习资料 分布式复习资料 第1章分布式系统是若干独立.........
分布式系统原理与泛型考博整理_教育学_高等教育_教育专区。分布式系统原理与范型 第二版 课后习题整理 第1 章 概述
分布式系统的另一个定义,它是各自独立的.........
tanenbaum maarten van steen chapter 5 na......
分布式操作系统课,分布式系统原理与范型课等都用的上分布式操作系统课,分布式系统原理与范型课等都用的上隐藏>> 分布式系统 2010春季学期期末考试北京大学计算机系,........
tanenbaum maarten van steen chapter 8 fa......
google云计算原理-分布式锁服务chubby_计算机软件及应用_it计算机_专业资料。分布式锁服务chubby 1 ? 主要用于解决分布式一致性问题 ? 在一个分布式系统中,有一组.........
分布式锁服务算法 9页 免费 google云计算原理-分布式... 54页 免费......
--我把它叫做时序控制锁 zookeeper的应用-分布式锁介绍core里的分布式锁的使用: 我们是利用zk实现了时序控制锁。 看代码演示和demo案例 如需要细细讲解zookeeper原理.........
《云计算》学习笔记3——google的云计算原理与应用(分布式锁服务——chubby)_互联网_it计算机_专业资料。一、分布式锁服务 今天,要接触有些难理解的知识点了,这.........
本文首先分析了分布式文件系统 l ust re 分布式锁管理器的基本原理 和实现......
redis实现分布式锁_计算机软件及应用_it计算机_专业资料。redis 实现分布式锁 ** * @author :blog.csdn.netjava2000_wl * @version 0 * public .........
集群运行原理及功能介绍 五、rhcs 集群运行原理及功能介绍 1、 分布式集群管理器...2、锁管理(dlm) distributed lock manager,简称 dlm,表示一个分布式锁管理器,.........
《分布式协调服务——zookeeper(原理与最佳实践)》倪超_互联网_it计算机_专业...名字服务 ? 分布式通知协调 ? 分布式锁 ? barriers ? 队列 ? 集群管理 三.........
分布式缓存服务器设计原理_计算机软件及应用_it计算机_专业资料。分布式缓存服务器设计原理 数据是如何被分布到多个服务器上的?(一致性哈 希算法)假设有 n 台.........
大规模分布式存储系统:原理解析与架构实战_互联网_it计算机_专业资料。大规模...存储系统 15 读取一个节点,如果对应的页面不在内存中,需要从磁盘中读取并缓存.........
memcached的原理及应用_it计算机_专业资料。memcached的原理及应用memcached详解及应用 part i memcached的基础 memcached的基础 memcached是高性能的分布式内存缓存服.........
王珊数据库原理8_计算机互联网_科技_ppt专区。王珊数据库原理,sql,数据库技术...分布式缓存器也是一个 瓶颈,可扩充性不佳 – 无共享资源结构不易做到负载平衡;.........
分布式系统原理与范型课后习题答案_研究生入学考试_高等教育_教育专区。第一章 绪论...答:可扩展性可以通过分布式、复制和缓存来获得。 8、多处理器系统与多计算机.........
关键词:分布式系统原理课后习题答案 12 相关文档推荐 分布式系统原理与范型课后...答:可扩展性可以通过分布式、复制和缓存来获得。 8、多处理器系统与多计算机.........
《分布式系统原理与范型 (第二版)》复习资料_工学_高等教育_教育专区。《...递归名称解析优点: 1) 与迭代名称解析相比,缓存结果更为有效 2) 减少通信开销.........中国领先的IT技术网站
51CTO旗下网站
分布式文件系统HDFS解读
HDFS是HadoopDistributedFileSystem的简称，既然是分布式文件系统，首先它必须是一个文件系统，那么在Hadoop上面的文件系统会不会也像一般的文件系统一样由目录结构和一组文件构成呢？分布式是不是就是将文件分成几部分分别存储在不同的机器上呢？看完本文的HDFS解读，你就会明白的。
作者：bearsorry来源：| 14:00
【前言】是蛮久木有写过关于hadoop的博客了额，虽然最近也看了一些关于linux的基础知识，但似乎把这个东西忘记了，其实时不时回顾一下以前的知识还是蛮有意思的，且行且忆！我们Hadoop主要由HDFS和MapReduce引擎两部分组成。最底部是HDFS，它存储Hadoop集群中所有存储节点上的文件。HDFS的上一层是MapReduce引擎，该引擎由JobTrackers和TaskTrackers组成。这篇博客就主要来讲讲HDFS吧~~~
HDFS是HadoopDistributedFileSystem的简称，既然是分布式文件系统，首先它必须是一个文件系统，那么在Hadoop上面的文件系统会不会也像一般的文件系统一样由目录结构和一组文件构成呢？分布式是不是就是将文件分成几部分分别存储在不同的机器上呢？HDFS到底有什么优点值得这么小题大作呢？
好吧，让我们带着疑问一个个去探索吧！
一、HDFS基本概念
HDFS默认的最基本的存储单位是64M的数据块，这个数据块可以理解和一般的文件里面的分块是一样的
2、元数据节点和数据节点
元数据节点（namenode）用来管理文件系统的命名空间，它将所有的文件和文件夹的元数据保存在一个文件系统树中。
数据节点（datanode）就是用来存储数据文件的。
从元数据节点（secondarynamenode）不是我们所想象的元数据节点的备用节点，其实它主要的功能是主要功能就是周期性将元数据节点的命名空间镜像文件和修改日志合并，以防日志文件过大。
这里先来弄清楚这个三种节点的关系吧！其实元数据节点上存储的东西就相当于一般文件系统中的目录，也是有命名空间的映射文件以及修改的日志，只是分布式文件系统就将数据分布在各个机器上进行存储罢了，下面你看看这几张说明图应该就能明白了！
Namenode与secondarynamenode之间的进行checkpoint的过程。
3、HDFS中的数据流
客户端(client)用FileSystem的open()函数打开文件，DistributedFileSystem用RPC调用元数据节点，得到文件的数据块信息。对于每一个数据块，元数据节点返回保存数据块的数据节点的地址。DistributedFileSystem返回FSDataInputStream给客户端，用来读取数据。客户端调用stream的read()函数开始读取数据。DFSInputStream连接保存此文件第一个数据块的最近的数据节点。Data从数据节点读到客户端(client)，当此数据块读取完毕时，DFSInputStream关闭和此数据节点的连接，然后连接此文件下一个数据块的最近的数据节点。当客户端读取完毕数据的时候，调用FSDataInputStream的close函数。
整个过程就是如图所示：
客户端调用create()来创建文件，DistributedFileSystem用RPC调用元数据节点，在文件系统的命名空间中创建一个新的文件。元数据节点首先确定文件原来不存在，并且客户端有创建文件的权限，然后创建新文件。DistributedFileSystem返回DFSOutputStream，客户端用于写数据。客户端开始写入数据，DFSOutputStream将数据分成块，写入dataqueue。Dataqueue由DataStreamer读取，并通知元数据节点分配数据节点，用来存储数据块(每块默认复制3块)。分配的数据节点放在一个pipeline里。DataStreamer将数据块写入pipeline中的第一个数据节点。第一个数据节点将数据块发送给第二个数据节点。第二个数据节点将数据发送给第三个数据节点。DFSOutputStream为发出去的数据块保存了ackqueue，等待pipeline中的数据节点告知数据已经写入成功。如果数据节点在写入的过程中失败：关闭pipeline，将ackqueue中的数据块放入dataqueue的开始。
整个过程如图所示：
二、HDFS构架与设计
Hadoop也是一个能够分布式处理大规模海量数据的软件框架，这一切都是在可靠、高效、可扩展的基础上。Hadoop的可靠性&&因为Hadoop假设计算元素和存储会出现故障，因为它维护多个工作数据副本，在出现故障时可以对失败的节点重新分布处理。Hadoop的高效性&&在MapReduce的思想下，Hadoop是并行工作的，以加快任务处理速度。Hadoop的可扩展&&依赖于部署Hadoop软件框架计算集群的规模，Hadoop的运算是可扩展的，具有处理PB级数据的能力。
Hadoop主要由HDFS（HadoopDistributedFileSystem）和MapReduce引擎两部分组成。最底部是HDFS，它存储Hadoop集群中所有存储节点上的文件。HDFS的上一层是MapReduce引擎，该引擎由JobTrackers和TaskTrackers组成。
HDFS可以执行的操作有创建、删除、移动或重命名文件等，架构类似于传统的分级文件系统。需要注意的是，HDFS的架构基于一组特定的节点而构建（参见图2），这是它自身的特点。HDFS包括唯一的NameNode，它在HDFS内部提供元数据服务；DataNode为HDFS提供存储块。由于NameNode是唯一的，这也是HDFS的一个弱点（单点失败）。一旦NameNode故障，后果可想而知。
1、HDFS构架（如图所示）
2、HDFS的设计
1）错误检测和快速、自动的恢复是HDFS的核心架构目标。
2）比之关注数据访问的低延迟问题，更关键的在于数据访问的高吞吐量。
3）HDFS应用对文件要求的是write-one-read-many访问模型。
4）移动计算的代价比之移动数据的代价低。
3、文件系统的namespace
Namenode维护文件系统的namespace，一切对namespace和文件属性进行修改的都会被namenode记录下来，连文件副本的数目称为replication因子，这个也是由namenode记录的。
4、数据复制
Namenode全权管理block的复制，它周期性地从集群中的每个Datanode接收心跳包和一个Blockreport。心跳包的接收表示该Datanode节点正常工作，而Blockreport包括了该Datanode上所有的block组成的列表。HDFS采用一种称为rack-aware的策略来改进数据的可靠性、有效性和网络带宽的利用。完成对副本的存放。
5、文件系统元数据的持久化
Namenode在内存中保存着整个文件系统namespace和文件Blockmap的映像。这个关键的元数据设计得很紧凑，因而一个带有4G内存的Namenode足够支撑海量的文件和目录。当Namenode启动时，它从硬盘中读取Editlog和FsImage，将所有Editlog中的事务作用（apply)在内存中的FsImage，并将这个新版本的FsImage从内存中flush到硬盘上,然后再truncate这个旧的Editlog，因为这个旧的Editlog的事务都已经作用在FsImage上了。这个过程称为checkpoint。在当前实现中，checkpoint只发生在Namenode启动时，在不久的将来我们将实现支持周期性的checkpoint。
6、通信协议
所有的HDFS通讯协议都是构建在TCP/IP协议上。客户端通过一个可配置的端口连接到Namenode，通过ClientProtocol与Namenode交互。而Datanode是使用DatanodeProtocol与Namenode交互。从ClientProtocol和Datanodeprotocol抽象出一个远程调用(RPC），在设计上，Namenode不会主动发起RPC，而是是响应来自客户端和Datanode的RPC请求。
HDFS不是这么简单就能说清楚的，在以后的博客中我还会继续研究hadoop的分布式文件系统，包括HDFS的源码分析等，现由于时间有限，暂时只做了以上一些简单的介绍吧，希望对大家由此对HDFS有一定的了解！
【编辑推荐】
【责任编辑： TEL：（010）】
大家都在看猜你喜欢
头条头条外电头条外电
24H热文一周话题本月最赞
讲师：0人学习过
讲师：0人学习过
讲师：11人学习过
精选博文论坛热帖下载排行
本书全面深入地介绍网络安全的配置与实现技术，包括系统管理、用户账户、病毒防御、灾难恢复、文件备份、安全策略、注册表等服务器安全，用...
订阅51CTO邮刊1324人阅读
Hadoop（96）
HDFS（54）
前言在Hadoop的HDFS启动的时候,不知道大家有没有注意到一个细节,一般都是先启动NameNode,然后再启动DataNode,细想一下,原因就很简单了,因为NameNode要维护元数据信息,而这些信息都是要等待后续启动的DataNode的情况汇报才能逐步构建的.然后之后通过保持心跳的形式进行block块映射关系的维护与更新.而今天的文章就以此方面,对这块流程做全面的分析.相关涉及类依旧需要介绍一下相关的涉及类,首先要有一个大概的了解.下面是主要的类:1.DataNode--数据节点类,这个和之前的数据节点描述符类又又有点不同,里面也定义了许多与数据节点相关的方法.2.NaemNode--名字节点类,注册信息的处理以及心跳包的处理都需要名字节点处理,名字节点的处理方法会调用FSNamesystem大系统中的方法.3.DatanodeCommand以及BlockCommand--数据节点命令类以及他的子类,block相关命令类,此类用于名字节点心跳回复命令给数据节点时用的.4.FSNamesystem和DatanodeDescriptor--附属类,这些类中的某些方法会在上述过程中被用到.OK,涉及的类的总数也不多,下面讲述第一个流程,节点注册,数据节点是如何在启动的时候注册到名字节点的呢.节点注册节点的注册是在数据节点启动之后发生的,首先进入main主方法public static void main(String args[]) {
secureMain(args, null);
}然后进入secureMainpublic static void secureMain(String [] args, SecureResources resources) {
StringUtils.startupShutdownMessage(DataNode.class, args, LOG);
DataNode datanode = createDataNode(args, null, resources);
if (datanode != null)
datanode.join();
} catch (Throwable e) {
LOG.error(StringUtils.stringifyException(e));
System.exit(-1);
} finally {
// We need to add System.exit here because either shutdown was called or
// some disk related conditions like volumes tolerated or volumes required
// condition was not met. Also, In secure mode, control will go to Jsvc and
// the process hangs without System.exit.
(&Exiting Datanode&);
System.exit(0);
}进入createDataNode方法,中间还会有1,2个方法,最终会调用到核心的runDatanodeDaemon()方法/** Start a single datanode daemon and wait for it to finish.
If this thread is specifically interrupted, it will stop waiting.
* 数据节点启动的核心方法
public static void runDatanodeDaemon(DataNode dn) throws IOException {
if (dn != null) {
//register datanode
//首先注册节点
dn.register();
//后续开启相应线程
dn.dataNodeThread = new Thread(dn, dnThreadName);
dn.dataNodeThread.setDaemon(true); // needed for JUnit testing
dn.dataNodeThread.start();
}于是就调用了register的方法/**
* Register datanode
* The datanode needs to register with the namenode on startup in order
* 1) to report which storage it is serving now and
* 2) to receive a registrationID
* issued by the namenode to recognize registered datanodes.
* @see FSNamesystem#registerDatanode(DatanodeRegistration)
* @throws IOException
* 数据节点的注册方法,会调用到namenode上的注册方法
private void register() throws IOException {
if (dnRegistration.getStorageID().equals(&&)) {
setNewStorageID(dnRegistration);
while(shouldRun) {
// reset name to machineName. Mainly for web interface.
dnRegistration.name = machineName + &:& + dnRegistration.getPort();
//调用namenode上的注册方法
dnRegistration = namenode.register(dnRegistration);
....然后这里,我们可以看到,注册方法实质上调用名字节点的注册方法.跟踪名字节点的同名法方法////////////////////////////////////////////////////////////////
// DatanodeProtocol
////////////////////////////////////////////////////////////////
* 名字节点的注册方法,调用的是FSNameSystem方法
public DatanodeRegistration register(DatanodeRegistration nodeReg
) throws IOException {
//首先做版本验证
verifyVersion(nodeReg.getVersion());
//调用namesystem的注册节点方法
namesystem.registerDatanode(nodeReg);
return nodeR
}名字节点调用的又是命名系统的方法，对注册节点的判断分为以下3种情况1、现有的节点进行新的存储ID注册2、现有节点的重复注册，由于集群已经保存有此信息，进行网络位置的更新即可3、从未注册过的节点，直接进行分配新的存储ID进行注册。具体方法判断如下，方法比较长/**
* Register Datanode.
* The purpose of registration is to identify whether the new datanode
* serves a new data storage, and will report new data block copies,
* which the namen or the datanode is a replacement
* node for the data storage that was previously served by a different
* or the same (in terms of host:port) datanode.
* The data storages are distinguished by their storageIDs. When a new
* data storage is reported the namenode issues a new unique storageID.
* Finally, the namenode returns its namespaceID as the registrationID
* for the datanodes.
* namespaceID is a persistent attribute of the name space.
* The registrationID is checked every time the datanode is communicating
* with the namenode.
* Datanodes with inappropriate registrationID are rejected.
* If the namenode stops, and then restarts it can restore its
* namespaceID and will continue serving the datanodes that has previously
* registered with the namenode without restarting the whole cluster.
* @see org.apache.hadoop.hdfs.server.datanode.DataNode#register()
* 名字节点实现数据节点的注册操作
public synchronized void registerDatanode(DatanodeRegistration nodeReg
) throws IOException {
String dnAddress = Server.getRemoteAddress();
if (dnAddress == null) {
// Mostly called inside an RPC.
// But if not, use address passed by the data-node.
dnAddress = nodeReg.getHost();
// check if the datanode is allowed to be connect to the namenode
if (!verifyNodeRegistration(nodeReg, dnAddress)) {
throw new DisallowedDatanodeException(nodeReg);
String hostName = nodeReg.getHost();
// update the datanode's name with ip:port
DatanodeID dnReg = new DatanodeID(dnAddress + &:& + nodeReg.getPort(),
nodeReg.getStorageID(),
nodeReg.getInfoPort(),
nodeReg.getIpcPort());
nodeReg.updateRegInfo(dnReg);
nodeReg.exportedKeys = getBlockKeys();
&BLOCK* NameSystem.registerDatanode: &
+ &node registration from & + nodeReg.getName()
+ & storage & + nodeReg.getStorageID());
//取出主机相关信息
DatanodeDescriptor nodeS = datanodeMap.get(nodeReg.getStorageID());
DatanodeDescriptor nodeN = host2DataNodeMap.getDatanodeByName(nodeReg.getName());
//判断此节点之前是否已经存在
if (nodeN != null && nodeN != nodeS) {
//此情况为数据节点存在,但是使用了新的存储ID
(&BLOCK* NameSystem.registerDatanode: &
+ &node from name: & + nodeN.getName());
// nodeN previously served a different data storage,
// which is not served by anybody anymore.
//移动掉旧的datanodeID信息
removeDatanode(nodeN);
// physically remove node from datanodeMap
//从物理层面的记录进行移除
wipeDatanode(nodeN);
//重复注册的情况
if (nodeS != null) {
if (nodeN == nodeS) {
// The same datanode has been just restarted to serve the same data
// storage. We do not need to remove old data blocks, the delta will
// be calculated on the next block report from the datanode
NameNode.stateChangeLog.debug(&BLOCK* NameSystem.registerDatanode: &
+ &node restarted.&);
// nodeS is found
/* The registering datanode is a replacement node for the existing
data storage, which from now on will be served by a new node.
If this message repeats, both nodes might have same storageID
by (insanely rare) random chance. User needs to restart one of the
nodes with its data cleared (or user can just remove the StorageID
value in &VERSION& file under the data directory of the datanode,
but this is might not work if VERSION file format has changed
( &BLOCK* NameSystem.registerDatanode: &
+ &node & + nodeS.getName()
+ & is replaced by & + nodeReg.getName() +
& with the same storageID & +
nodeReg.getStorageID());
// update cluster map
//更新集群的网络信息
clusterMap.remove(nodeS);
nodeS.updateRegInfo(nodeReg);
nodeS.setHostName(hostName);
// resolve network location
resolveNetworkLocation(nodeS);
clusterMap.add(nodeS);
// also treat the registration message as a heartbeat
synchronized(heartbeats) {
if( !heartbeats.contains(nodeS)) {
heartbeats.add(nodeS);
//update its timestamp
nodeS.updateHeartbeat(0L, 0L, 0L, 0);
nodeS.isAlive =
// this is a new datanode serving a new data storage
//当此时确认为一个新的节点时,为新节点分配存储ID
if (nodeReg.getStorageID().equals(&&)) {
// this data storage has never been registered
// it is either empty or was created by pre-storageID version of DFS
nodeReg.storageID = newStorageID();
NameNode.stateChangeLog.debug(
&BLOCK* NameSystem.registerDatanode: &
+ &new storageID & + nodeReg.getStorageID() + & assigned.&);
// register new datanode
//创建新的节点
DatanodeDescriptor nodeDescr
= new DatanodeDescriptor(nodeReg, NetworkTopology.DEFAULT_RACK, hostName);
resolveNetworkLocation(nodeDescr);
unprotectedAddDatanode(nodeDescr);
clusterMap.add(nodeDescr);
// also treat the registration message as a heartbeat
//将注册信息加入到心跳
synchronized(heartbeats) {
heartbeats.add(nodeDescr);
nodeDescr.isAlive =
// no need to update its timestamp
// because its is done when the descriptor is created
}移除原有信息的代码如下，还做了更新操作/**
* remove a datanode descriptor
* @param nodeInfo datanode descriptor
private void removeDatanode(DatanodeDescriptor nodeInfo) {
synchronized (heartbeats) {
if (nodeInfo.isAlive) {
//更新集群中的统计信息
updateStats(nodeInfo, false);
//从心跳列表信息中移除对于此节点的心跳信息
heartbeats.remove(nodeInfo);
nodeInfo.isAlive =
//移除第二关系中数据块对于此节点的映射关系
for (Iterator&Block& it = nodeInfo.getBlockIterator(); it.hasNext();) {
removeStoredBlock(it.next(), nodeInfo);
unprotectedRemoveDatanode(nodeInfo);
//从集群图中移除此节点信息
clusterMap.remove(nodeInfo);
}心跳机制心跳机制最简单的由来就是为了证明数据节点还活着，如果一段时间内datanode没有向namenode发送心跳包信息，就会被dead状态。并且datanode从心跳包回复中获取命令信息，然后进行下一步操作，所以从这里可以看出，心跳机制在整个HDFS系统中都有很重要的作用。下面一步步揭开HDFS心跳机制的实现。首先心跳信息是由数据节点发起的，主动方在Datanode上，就是下面这个方法/**
* Main loop for the DataNode.
Runs until shutdown,
* forever calling remote NameNode functions.
* datanode在循环中不断向名字节点发送心跳信息
public void offerService() throws Exception {
(&using BLOCKREPORT_INTERVAL of & + blockReportInterval + &msec& +
& Initial delay: & + initialBlockReportDelay + &msec&);
// Now loop for a long time....
while (shouldRun) {
long startTime = now();
// Every so often, send heartbeat or block-report
if (startTime - lastHeartbeat & heartBeatInterval) {
// All heartbeat messages include following info:
// -- Datanode name
// -- data transfer port
// -- Total capacity
// -- Bytes remaining
//向名字节点发送此时节点的一些信息，dfs使用量，剩余使用量信息等
lastHeartbeat = startT
//调用namenode.sendHeartbeat进行心跳信息的发送，返回数据节点的操作命令
DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
data.getCapacity(),
data.getDfsUsed(),
data.getRemaining(),
xmitsInProgress.get(),
getXceiverCount());
myMetrics.addHeartBeat(now() - startTime);
//(&Just sent heartbeat, with name & + localName);
//进行返回命令的处理，如果没有成功不进行后续block块上报工作
if (!processCommand(cmds))
}首先先观察前部分代码，可以看到，这是一个循环，而且是周期性的发送消息，然后调用namenode的方法进行心跳信息发送，然后接收DatanodeCommand回复命令，然后再本节点执行。然后，跟踪一下里面namenode的方法/**
* Data node notify the name node that it is alive
* Return an array of block-oriented commands for the datanode to execute.
* This will be either a transfer or a delete operation.
* 数据节点调用此方法进行心跳信息的发送
public DatanodeCommand[] sendHeartbeat(DatanodeRegistration nodeReg,
long capacity,
long dfsUsed,
long remaining,
int xmitsInProgress,
int xceiverCount) throws IOException {
//对节点注册信息的确认
verifyRequest(nodeReg);
//然后再次调用fsnamesystem的handleHeartbeat方法
return namesystem.handleHeartbeat(nodeReg, capacity, dfsUsed, remaining,
xceiverCount, xmitsInProgress);
}namenode调用的又是命名系统的方法/**
* The given node has reported in.
This method should:
* 1) Record the heartbeat, so the datanode isn't timed out
* 2) Adjust usage stats for future block allocation
* If a substantial amount of time passed since the last datanode
* heartbeat then request an immediate block report.
* @return an array of datanode commands
* @throws IOException
* 一个给定的数据节点进行心跳信息的上报，主要做2个操作
* 1.心跳信息的记录，避免数据节点超时
* 2.调整新的名字节点中维护的数据块分配情况
DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
long capacity, long dfsUsed, long remaining,
int xceiverCount, int xmitsInProgress) throws IOException {
DatanodeCommand cmd =
synchronized (heartbeats) {
synchronized (datanodeMap) {
DatanodeDescriptor nodeinfo =
nodeinfo = getDatanode(nodeReg);
} catch(UnregisteredDatanodeException e) {
return new DatanodeCommand[]{DatanodeCommand.REGISTER};
// Check if this datanode should actually be shutdown instead.
if (nodeinfo != null && shouldNodeShutdown(nodeinfo)) {
setDatanodeDead(nodeinfo);
throw new DisallowedDatanodeException(nodeinfo);
//如果不存在此节点信息，说明此节点还未注册，返回节点注册命令
if (nodeinfo == null || !nodeinfo.isAlive) {
return new DatanodeCommand[]{DatanodeCommand.REGISTER};
updateStats(nodeinfo, false);
nodeinfo.updateHeartbeat(capacity, dfsUsed, remaining, xceiverCount);
updateStats(nodeinfo, true);看前半段的代码分析，主要是做相应节点的信息更新，如果节点是未注册过的野节点的话，返回注册命令，但是还有下半段的操作，/**
* The given node has reported in.
This method should:
* 1) Record the heartbeat, so the datanode isn't timed out
* 2) Adjust usage stats for future block allocation
* If a substantial amount of time passed since the last datanode
* heartbeat then request an immediate block report.
* @return an array of datanode commands
* @throws IOException
* 一个给定的数据节点进行心跳信息的上报，主要做2个操作
* 1.心跳信息的记录，避免数据节点超时
* 2.调整新的名字节点中维护的数据块分配情况
DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
long capacity, long dfsUsed, long remaining,
int xceiverCount, int xmitsInProgress) throws IOException {
...........
//check lease recovery
//检查租约过期情况
cmd = nodeinfo.getLeaseRecoveryCommand(Integer.MAX_VALUE);
if (cmd != null) {
return new DatanodeCommand[] {cmd};
//新建命令集合，心跳回复将返回许多的命令
ArrayList&DatanodeCommand& cmds = new ArrayList&DatanodeCommand&();
//check pending replication
//获取待复制的副本块命令，命令中包含有副本块列表内容
cmd = nodeinfo.getReplicationCommand(
maxReplicationStreams - xmitsInProgress);
if (cmd != null) {
cmds.add(cmd);
//check block invalidation
//块删除命令
cmd = nodeinfo.getInvalidateBlocks(blockInvalidateLimit);
if (cmd != null) {
cmds.add(cmd);
// check access key update
if (isAccessTokenEnabled && nodeinfo.needKeyUpdate) {
cmds.add(new KeyUpdateCommand(accessTokenHandler.exportKeys()));
nodeinfo.needKeyUpdate =
// check for balancer bandwidth update
if (nodeinfo.getBalancerBandwidth() & 0) {
cmds.add(new BalancerBandwidthCommand(nodeinfo.getBalancerBandwidth()));
// set back to 0 to indicate that datanode has been sent the new value
nodeinfo.setBalancerBandwidth(0);
if (!cmds.isEmpty()) {
//返回命令组
return cmds.toArray(new DatanodeCommand[cmds.size()]);
}就是命名系统类会检查节点block情况，进行命令回复，比如添加需要复制的block请求，无效快的删除请求等等。下面是一个示例//与block命令相关的函数
BlockCommand getReplicationCommand(int maxTransfers) {
//获取待复制的副本block块列表
List&BlockTargetPair& blocktargetlist = replicateBlocks.poll(maxTransfers);
//将变量保证在BlockCommand中进行返回
return blocktargetlist == null? null:
new BlockCommand(DatanodeProtocol.DNA_TRANSFER, blocktargetlist);
}因为BlockCommand是DataCommand的子集，也是属于DatanodeCommand。在这里有必要了解一下，datanode执行命令的形式，首先是父类，比较简单一些//DatanodeCommand继承自Writable序列化类，说明命令是被序列化传输的
public abstract class DatanodeCommand implements Writable {
static class Register extends DatanodeCommand {
private Register() {super(DatanodeProtocol.DNA_REGISTER);}
public void readFields(DataInput in) {}
public void write(DataOutput out) {}
static class Finalize extends DatanodeCommand {
private Finalize() {super(DatanodeProtocol.DNA_FINALIZE);}
public void readFields(DataInput in) {}
public void write(DataOutput out) {}
//action保存了命令操作类型
///////////////////////////////////////////
// Writable
///////////////////////////////////////////
//DatanodeCommand将命令操作写在序列化流中
public void write(DataOutput out) throws IOException {
out.writeInt(this.action);
public void readFields(DataInput in) throws IOException {
this.action = in.readInt();
}这里的action很重要，保存的就是命令的类型，同时，注意这些都是可序列化的，用于RPC传输。那么与Block相关的命令在上面做了哪些的改变才使得能把block块列表的信息也传入命令中呢，答案就在下面/****************************************************
* A BlockCommand is an instruction to a datanode
* regarding some blocks under its control.
* the DataNode to either invalidate a set of indicated
* blocks, or to copy a set of indicated blocks to
* another DataNode.
****************************************************/
public class BlockCommand extends DatanodeCommand {
Block blocks[];
DatanodeInfo targets[][];
public BlockCommand() {}
* Create BlockCommand for transferring blocks to another datanode
* @param blocktargetlist
blocks to be transferred
public BlockCommand(int action, List&BlockTargetPair& blocktargetlist) {
super(action);
blocks = new Block[blocktargetlist.size()];
targets = new DatanodeInfo[blocks.length][];
for(int i = 0; i & blocks. i++) {
BlockTargetPair p = blocktargetlist.get(i);
blocks[i] = p.
targets[i] = p.
private static final DatanodeInfo[][] EMPTY_TARGET = {};
* Create BlockCommand for the given action
* @param blocks blocks related to the action
public BlockCommand(int action, Block blocks[]) {
super(action);
this.blocks =
this.targets = EMPTY_TARGET;
//重载序列化写入方法
public void write(DataOutput out) throws IOException {
super.write(out);
out.writeInt(blocks.length);
//将block块依次序列化写入
for (int i = 0; i & blocks. i++) {
blocks[i].write(out);
out.writeInt(targets.length);
for (int i = 0; i & targets. i++) {
out.writeInt(targets[i].length);
for (int j = 0; j & targets[i]. j++) {
targets[i][j].write(out);
OK，心跳回复命令已经清楚，看看数据节点如何执行命名节点返回的心跳命令然后做出调整。/**
* Process an array of datanode commands
* @param cmds an array of datanode commands
* @return true if further processing may be required or false otherwise.
* 数据节点批量执行操作
private boolean processCommand(DatanodeCommand[] cmds) {
if (cmds != null) {
for (DatanodeCommand cmd : cmds) {
//在命令组中，只要有一条命令执行出错，整个执行过程就算失败
if (processCommand(cmd) == false) {
} catch (IOException ioe) {
LOG.warn(&Error processing datanode Command&, ioe);
}然后是单条命令执行，这才是我们想看到的/**
* @param cmd
* @return true if further processing may be required or false otherwise.
* @throws IOException
* 调用单条命令处理方法
private boolean processCommand(DatanodeCommand cmd) throws IOException {
if (cmd == null)
final BlockCommand bcmd = cmd instanceof BlockCommand? (BlockCommand)cmd:
//取出命令的action值类型，进行分别判断处理
switch(cmd.getAction()) {
case DatanodeProtocol.DNA_TRANSFER:
// Send a copy of a block to another datanode
case DatanodeProtocol.DNA_INVALIDATE:
//如果是无效块，则进行blockScanner类扫描删除操作
// Some local block(s) are obsolete and can be
// safely garbage-collected.
Block toDelete[] = bcmd.getBlocks();
if (blockScanner != null) {
blockScanner.deleteBlocks(toDelete);
data.invalidate(toDelete);
} catch(IOException e) {
checkDiskError();
myMetrics.incrBlocksRemoved(toDelete.length);
case DatanodeProtocol.DNA_SHUTDOWN:
case DatanodeProtocol.DNA_REGISTER:
//如果是注册命令，则调用注册操作
case DatanodeProtocol.DNA_FINALIZE:
storage.finalizeUpgrade();
case UpgradeCommand.UC_ACTION_START_UPGRADE:
// start distributed upgrade here
processDistributedUpgradeCommand((UpgradeCommand)cmd);
case DatanodeProtocol.DNA_RECOVERBLOCK:
recoverBlocks(bcmd.getBlocks(), bcmd.getTargets());
case DatanodeProtocol.DNA_ACCESSKEYUPDATE:
(&DatanodeCommand action: DNA_ACCESSKEYUPDATE&);
if (isBlockTokenEnabled) {
blockTokenSecretManager.setKeys(((KeyUpdateCommand) cmd).getExportedKeys());
case DatanodeProtocol.DNA_BALANCERBANDWIDTHUPDATE:
LOG.warn(&Unknown DatanodeCommand action: & + cmd.getAction());
}对每种命令，取出action值类别，进行分别处理。OK，到这里就处理完了前半部分代码所做的事情了，在后面会进行数据块上报的工作，准确的说，这方面的操作就不算是心跳机制里面的过程了，但是都是在一个大循环中进行的。//检测新接收到的block
// check if there are newly received blocks
Block [] blockArray=
String [] delHintArray=
synchronized(receivedBlockList) {
synchronized(delHints) {
int numBlocks = receivedBlockList.size();
if (numBlocks & 0) {
if(numBlocks!=delHints.size()) {
LOG.warn(&Panic: receiveBlockList and delHints are not of the same length& );
// Send newly-received blockids to namenode
blockArray = receivedBlockList.toArray(new Block[numBlocks]);
delHintArray = delHints.toArray(new String[numBlocks]);
if (blockArray != null) {
if(delHintArray == null || delHintArray.length != blockArray.length ) {
LOG.warn(&Panic: block array & delHintArray are not the same& );
//将接收到的新block信息上报
namenode.blockReceived(dnRegistration, blockArray, delHintArray);
synchronized (receivedBlockList) {
synchronized (delHints) {
for(int i=0; i&blockArray. i++) {
receivedBlockList.remove(blockArray[i]);
delHints.remove(delHintArray[i]);
}数据节点进行心跳信息上传，名字节点如何判断节点是否长期没有上报心跳呢，在这之间必然会有监控线程，在命名系统类中，相关的变量/**
* Stores a set of DatanodeDescriptor objects.
* This is a subset of {@link #datanodeMap}, containing nodes that are
* considered alive.
* The {@link HeartbeatMonitor} periodically checks for outdated entries,
* and removes them from the list.
* 是datanodeMap的子集，保存了节点中存活的节点集合，在HeartbeatMonitor中会被使用
ArrayList&DatanodeDescriptor& heartbeats = new ArrayList&DatanodeDescriptor&();此变量就代表着现有存活的节点列表/**
* Periodically calls heartbeatCheck() and updateAccessKey()
* 心跳监控线程
class HeartbeatMonitor implements Runnable {
//上次心跳的检测时间
private long lastHeartbeatC
private long lastAccessKeyU
public void run() {
while (fsRunning) {
long now = now();
if (lastHeartbeatCheck + heartbeatRecheckInterval & now) {
//如果在间隔时间内，做心跳检测
heartbeatCheck();
lastHeartbeatCheck =
if (isAccessTokenEnabled && (lastAccessKeyUpdate + accessKeyUpdateInterval & now)) {
updateAccessKey();
lastAccessKeyUpdate =
} catch (Exception e) {
FSNamesystem.LOG.err上面是监控线程的执行方法，在做心跳检测的方法时，作者采用了一种很有远见的想法，采用了故障寻找，故障分离的办法，原因是考虑到分布式系统的负载情况，以及对于系统整体性能的影响，下面是故障寻找/**
* Check if there are any expired heartbeats, and if so,
* whether any blocks have to be re-replicated.
* While removing dead datanodes, make sure that only one datanode is marked
* dead at a time within the synchronized section. Otherwise, a cascading
* effect causes more datanodes to be declared dead.
* 检测是否有任何过期的心跳，移除dead状态的节点
void heartbeatCheck() {
//安全模式下不做任何的检查
if (isInSafeMode()) {
// not to check dead nodes if in safemode
boolean allAlive =
while (!allAlive) {
boolean foundDead =
DatanodeID nodeID =
// locate the first dead node.
//定位寻找第一个dead故障状态的节点
synchronized(heartbeats) {
for (Iterator&DatanodeDescriptor& it = heartbeats.iterator();
it.hasNext();) {
DatanodeDescriptor nodeInfo = it.next();
if (isDatanodeDead(nodeInfo)) {
//一旦找到，并取出对应消息并跳出循环
foundDead =
nodeID = nodeI
然后是处理，但是处理之前还需要在判断一次是否是故障节点，分布式的环境嘛，各种因素影响。// acquire the fsnamesystem lock, and then remove the dead node.
if (foundDead) {
//为了确保同步性，进行加锁操作，进行dead故障状态节点的处理操作
synchronized (this) {
synchronized(heartbeats) {
synchronized (datanodeMap) {
DatanodeDescriptor nodeInfo =
nodeInfo = getDatanode(nodeID);
} catch (IOException e) {
nodeInfo =
if (nodeInfo != null && isDatanodeDead(nodeInfo)) {
(&BLOCK* NameSystem.heartbeatCheck: &
+ &lost heartbeat from & + nodeInfo.getName());
//移除此节点的信息，此方法会调用headtbeat的移除nodeinfo操作
removeDatanode(nodeInfo);
//重置标志位
allAlive = !foundD
}心跳故障处理检测完毕，这种故障处理检测的思想或许对于我们日后自己动手设计大型分布式系统可能会有所启发。全部代码的分析请点击链接,后续将会继续更新HDFS其他方面的代码分析。参考文献《Hadoop技术内部–HDFS结构设计与实现原理》.蔡斌等
参考知识库
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：725735次
积分：8028
积分：8028
排名：第1923名
原创：236篇
评论：214条
毕业于HDU计算机系，研究领域分布式计算，大数据，数据挖掘，机器学习，算法，Hadoop开源社区活跃贡献者,其中主要研究HDFS.曾就职于国内女性电商平台蘑菇街，目前就职于唯品会上海研发中心，数据平台与应用部门
文章：29篇
阅读：100398
阅读：8351
阅读：11531
文章：11篇
阅读：16517
文章：36篇
阅读：92826
(3)(3)(4)(7)(6)(3)(2)(5)(4)(4)(6)(6)(4)(5)(4)(6)(9)(6)(4)(5)(6)(10)(11)(23)(20)(26)(37)(5)(4)(1)