前言

我们知道 HDFS 文件是 write-once-read-many，并且不支持客户端的并行写操作，那么这里就需要一种机制保证对 HDFS 文件的互斥操作。HDFS 提供了租约（Lease）机制来实现这个功能，租约是 HDFS 中一个很重要的概念，是 Namenode 给予租约持有者（LeaseHolder，一般是客户端）在规定时间内拥有文件权限（写文件）的合同。

在 HDFS 中，客户端写文件时需要先从租约管理器(LeaseManager）申请一个租约，成功申请租约之后客户端就成为了租约持有者，也就拥有了对该 HDFS 文件的独占权限，其他客户端在该租约有效时无法打开这个 HDFS 文件进行操作。

Namenode 的租约管理器保存了 HDFS 文件与租约、租约与租约持有者的对应关系，租约管理器还会定期检查它维护的所有租约是否过期。租约管理器会强制收回过期的租约，所以租约持有者需要定期更新租约 （renew)，维护对该文件的独占锁定。当客户端完成了对文件的写操作，关闭文件时，必须在租约管理器中释放租约。

LeaseManager.Lease

我们知道一个 HDFS 客户端是可以同时打开多个 HDFS 文件进行读写操作的，为了便于管理，在租约管理器中将一个客户端打开的所有文件组织在一起构成一条记录，也就是 LeaseManager. Lease 类。

给出了 Lease 类定义的所有字段，其中 holder 字段保存了客户端也就是租约持有者的信息，paths 字段保存了该客户端打开的所有 HDFS 文件的路径，lastUpdate 字段则保存了租约最后的更新时间。

Lease 类中还有三个比较重要的方法需要重点讲解，这些方法都是供 LeaseManager 管理 Lease 时调用的

• renew：renew方法用于更新客户端的 lastUpdate 最近更新时间。

• expiredSoftLimit： 用于判附当前柤约是否超出了软限制(sofLimit)，软限制是写文件规定的租约超时时间，默认是60 秒，不可以配置。

• expiredHardLimit：用于判断当前租约是否超出了硬限制 (hardLimit)，硬限制是用于考虑文件关闭异常时，强制回收租约的时间，默认是 60 分钟，不可以配置。在 LeaseManager 中有一个内部类用于定期检查租约的更新情况，当超过硬限制时间时，会触发租约恢复机制。

LeaseManager

LeaseManager 是 Namenode 中维护所有租约操作的类，它不仅仅保存了 HDFS 中所有租约的信息，提供租约的增、删、改、查方法，同时还维护了一个 Monitor 线程定期检查租约是否超时，对于长时间没有更新租约的文件（超过硬限制时间)，LeaseManager 会触发租约恢复机制，然后关闭文件。

在 LeaseManager 中使用数据结构 leases、 sortedLeases 以及 sortedLeasesByPath 三个字段保存 Namenode 中的所有租约；使用 Imthread 字段保存租约检查线程；使用 sofLimit 字段保存软限制时间（默认是60 秒，不可以配置)；使用 hadrLimit 字段保存硬限制时间（默认是60分钟，不可以配置）。

下面我们看一下 LeaseManager 保存租约的三个字段：leases 、sortedLeases 和 sortedLeasesByPath。

• leases：保存了租约持有者与租约的对应关系。

• sortedLeases：以租约更新时间为顺序保存 LeaseManager 中的所有租约，如果更新时间相同，则按照租约持有者的字典序保存。

• sortedLeasesByPath：保存了文件路径与租约的对应关系，以路径的字典序为顺序保存。

了解了 LeaseManager 定义的字段，下面我们学习 LeaseManager 提供的增加、删除、更新以及恢复租约的方法。

添加租约 - addLease

当客户端创建文件和追加写文件时，FSNamesystem.startFilelnternal() 以及 appendFilelnternal() 方法都会调用 LeaseManager.addLease() 为该客户端在 HDFS 文件上添加一个租约。addLease() 方法有两个参数，其中 holder 参数保存租约的持有者信息，src 参数则保存创建或者追加写文件的路径，这两个参数分别对应于 ClientProtocol.ercate() 或者 append() 方法中的 clientName 和 src 参数。addLease() 方法的实现也非常简单，先是通过 getLease() 方法构造租约，然后在 LeaseManager 定义的保存租约的数据结构中添加这个租约的信息。

synchronized Lease addLease(String holder, long inodeId) {
    Lease lease = getLease(holder);
    if (lease == null) {
      // 构造 Lease 对象
      lease = new Lease(holder);
      // 在leaseManager.leases字段添加Lease对象
      leases.put(holder, lease);
    } else {
      renewLease(lease);
    }
    leasesById.put(inodeId, lease);
    lease.files.add(inodeId);
    return lease;
  }

addLease() 在另外两种情况下也会被调用，就是 Namenode 读取 fsimage 文件时，fsimage 文件记录了当前 HDFS 文件处于构建状态中，这时需要重建这个构建中的文件并将文件对应的 INode 对象加入文件系统目录树中，然后还需要在 LeaseManager 中添加租约信息；以及在 Namenode 读取 editlog 时，editlog 记录了一个 OP_ADD 操作，也就是创建文件的操作， Namenode 创建完 INode 对象并添加到文件系统目录树之后，还需要在 LeaseManager 中添加租约信息。

租约检查 - FsNamesystem.checkLease

如图 3-51 所示，当客户端已经成功打开了一个 HDFS 文件并添加租约后，客户端调用 abandonBlock()放弃新申请的数据块(客户端数据流管道建立失败时)，调用 getAddtionalBlock() 申请新的数据块（客户端完成了上一个数据块的写入，申请新的数据块时），调用 completeFilelnternal()提交文件（客户端完成 HDFS 文件的写操作，提交文件时）等情况时， 都需要检查租约是否正常。

这里的检查操作是由 FSNamesystem.checkLease() 方法执行的，checkLease() 方法会检查当前 HDFS 文件是否存在、INode 是否是一个目录、文件是否处于构建中状态、文件是否已被删除、输入的文件的租约持有者与实际的租约持有者是否相同。如果这些检查不正常，则抛出 LeaseExpiredException。

租约更新 - renewLease

当客户端打开了一个文件用于写或者追加写操作时，LeaseManager 会保存这个客户端在该文件上的租约。客户端会启动一个 LeaseRenewer 定期更新租约，以防止租约过期。

租约更新操作是由 FSNamesystem.renewLease() 响应的，这个方法最终会调用 LeaseManager.renewLease() 方法。renewLease() 方法会首先从 sortedLeases 字段中移除这个租约，然后更新这个租约的最后更新时间，再重新加入 sortedLeases 中。这么做的原因是， sortedLeases 是一个以最后更新时间排序的集合，所以每次更新租约后，sortedLeases 中的顺序也需要重新改变。

删除租约 - removeLease

LeaseManager 中的租约会在两种情况下被删除。

• Namenode 关闭构建中的 HDFS 文件时，会调用 FSNamesystem.finalizeINodeFileUnderConstruction() 方法将 INode 从构建状态转换成非构建状态，同时由于客户端已经完成了文件的写操作，所以需要从 LeaseManager 中删除该文件的租约，这里调用了 removeLease() 方法删除租约。

• 在进行目录树的删除操作时，对于已经打开的文件，如果客户端从文件系统目录树中移出该 HDFS 文件，则会调用 removeLeaseWithPrefixPath()方法从 LeaseManager 中删除租约。

removeLease() 和 removeLeaseWithPrefixPath() 的实现都比较简单，从 LeaseManager 保存租约的数据结构中删除租约信息即可，读者可以直接参考以下代码：

租约检查 - Monitor 线程

我们知道租约管理器除了对租约提供增、删、改、查等操作外，还会定期检查所有租约， 对于长时间没有进行租约更新的文件，LeaseManager 会对这个文件进行租约恢复操作，然后关闭这个文件。在什么情况下会出现租约过期呢？我们知道 HDFS是一个分布式系统，客户端很有可能在打开一个文件之后出现故障，这也就造成了客户端不能完成租约更新以及写文件之后的租约删除操作，这时就会造成租约过期。

租约的定期检查操作是由 LeaseManager 的内部类 Monitor 执行的，Monitor 是一个线程类， 它的 run() 方法会每隔2 秒调用一次 LeaserManager.checkLeases() 方法检查租约。

在前面的小节中我们己经介绍过，LeaseManager 中有两个限制时间，其中软限制时间用于记录写文件规定的租约超时时间；硬限制时间则用于判断文件是否由于异常而未能正确关闭。在 checkLeases() 方法中就是使用硬限制时间（60 分钟）判断是否需要进行租约恢复操作的。

checkLeases() 方法会遍历 leaseManager 中管理的所有租约，找出所有超过硬限制时间而未更新的租约。由于租约保存了这个客户端打开的所有 HDFS 文件，所以 checkLeases() 方法会遍历这个租约上的所有文件，并调用 FSNamesystem.internalReleaseLease() 方法进行租约恢复操作。checkLeases() 方法的代码如下：

租约恢复 - Monitor 线程发起

对于 HDFS文件的租约恢复操作是通过调用 FSNamesystem.internalReleaseLease() 实现的， 这个方法用于将一个已经打开的文件进行租约恢复并关闭。如果成功关闭了文件， internalReleaseLease() 方法会返回 true：如果仅触发了租约恢复操作，则返回 false。我们知道租约恢复是针对已经打开的构建中的文件的，所以 internalReleaseLease() 会判断文件中所有数据块的状态，对于异常的状态则直接抛出异常。在 checkLeases() 方法中，对于调用 FSNamesystem.internalReleaseLease() 方法时抛出异常的租约，则直接调用 removeLease() 方法删除。

当文件处于构建状态时，有三种情况可以直接关闭文件，并返回 true。

• 这个文件所拥有的所有数据块都处于 COMPLETED 状态，也就是客户端还没有来得及关闭文件和释放租约就出现了故障，这时 interalReleaseLease() 可以直接调用 finalizeINodeFileUnderConstruction() 方法关闭文件并删除租约。

• 文件的最后一个数据块处于提交状态(COMMITTED)，并且该数据块至少有一个有效的副本，这时可以直接调用 finalizeINodeFileUnderConstruction() 方法关闭文件并删除租约。

• 文件的最后一个数据块处于构建中状态，但这个数据块的长度为 0，且当前没有 Datanode 汇报接收了这个数据块，这种情况很可能是客户端向数据流管道中写数据前发生了故障，这时可以将最后一个未写入数据的数据块删除，之后调用 finalizeINodeFileUnderConstruction() 方法关闭文件并删除租约。

当最后一个数据块处于 UNDER RECOVERY 或者 UNDER CONSTRUCTION 状态，且这个数据块己经写入了数据时，则构造一个新的时间戳作为 recoveryld，调用 initializeBlockRecovery() 触发租约恢复，更新当前文件的租约持有者为 “HDFS NameNode”。 internalReleaseLease() 的代码如下：

下面我们看一下租约恢复的过程，这里是在需要进行租约恢复的数据块上调用 initializeBlockRecovery() 方法，该方法会遍历所有保存副本的数据节点，选取一个最近一次进行汇报的数据节点作为主恢复节点，然后向这个数据节点发送租约恢复指令，Namenode 会通过心跳将租约恢复的名字节点指令下发给该恢复节点。

租约恢复指令是通过心跳响应携带给主恢复数据节点的，主恢复数据节点的租约恢复流程如图 3-52 所示。主恢复数据节点接收到指令后，会调用 Datanode.recoverBlock() 方法开始租约恢复，这个方法首先会通过 Inter DatanodeProtocolinitReplicaRecovery() 方法向数据流管道中参与租约恢复的数据节点收集副本信息，副本信息会以 ReplicaRecoveryinfo 对象返回给主恢复节点。initReplicaRecovery() 方法会从该数据块的所有副本中选取一个最好的状态，作为所有副本恢复的目标状态。然后主恢复节点会调用 InterDatanodeProtocol.updateReplicaUnderRecovery() 方法同步所有 Datanode 上该数据块副本至目标状态。同步结束后，这些数据节点上的副本长度和时间戳将一致。最后，主恢复节点会调用 DatanodeProtocol.commitBlock Synchronization() 向名字节点报告这次租约恢复的结果。

现在我们看一下数据块同步的提交方法 commitBlockSynchronization() 。 commitBlockSynchronization() 方法用于将进行了租约恢复的数据节点上的副本信息与名字节点上的数据块信息同步。这个方法的参数也比较多，其中 lastblock 是被恢复的数据块，newgenerationstamp 是租约恢复以后新的时间戳，newlength 是租约恢复以后副本的长度，closeFile 用于指示是否关闭数据块对应的 HDFS文件，deleteblock 用于指示是否直接删除这个数据块，newtargets 存储了租约恢复后保存这个数据块副本的数据节点列表，newtargetstorages 保存了数据节点的存储信,息。

在数据块恢复过程中，如果发现数据流管道中并不存在这个副本或者所有副本的长度都为0，则可以不用进行租约恢复操作，直接删除这个数据块即可，这时主恢复节点会将 deleteblock 字段设置为 ture, commitBlockSynchronization() 方法会从 Namenode 中删除这个数据块，然后关闭文件；否则，commitBlockSynchronization() 方法进行数据块更新操作，更新 Namenode 上数据块的时间戳和长度，使 Namenode 上的数据块信息与 Datanode 上进行租约恢复后的副本一致。由于可能只有部分数据节点参与租约恢复，所以还需要更新 DatanodeStoragelnfo 上的数据块信息，以及 INodeFile 中的数据块信息。
完成了 commitBlockSynchronization() 方法，Namenode 上数据块的信息与 Datanode 上的副本信息也就一致了，这时整个租约恢复流程也就结束了。

租约恢复 - 其他方式发起

租约恢复除了可以由 LeaseManager.Monitor 线程发起外，如图3-53 所示，还有以下三种情况会调用 FSNamesystem.recoverLeaselnternal() 方法触发租约恢复操作，这三种情况调用 recoverLeaselnternal() 的不同之处在于 force 字段的赋值不同。
客户端通过 ClientProtocol.recoverLease() 远程方法发起租约恢复，最终会由 FSNamesystem.recoverLeaselnternal() 响应，这里会将 force 字段置为 true，也就是强制执行 internalReleaseLease() 方法关闭文件并释放租约，而不用判断租约是否超过了软限制时间。

客户端通过 startFilelnternal() 打开一个文件以进行写操作，这时候会调用 recoverLeaselinternal() 方法检查是否有别的客户端打开了这个文件，以防止多个客户端同时写这个文件。如果有别的客户端打开了这个文件，recoverLeaseInternal() 方法会抛出 AlreadyBeingCreatedException 异常。这里的 force 字段设置为 false，方法会判断原租约持有者是否已经软超时(softLimit)，如果超时则进行租约恢复操作释放租约并关闭文件，为文件写操作做准备。

客户端通过 appendFilelnternal() 打开文件进行追加写操作与 startFilelnternal()是同一个道理，调用 recoverLeaselnternal() 方法检查是否有别的客户端同时打开了这个文件。

本文内容结合自 《Hadoop 2.X HDFS 源码剖析》以及自己的理解

希望对正在查看文章的您有所帮助，记得关注、评论、收藏，谢谢您