[TOC]
Namenode 元数据管理源码分析
本文讲解 Namenode 元数据管理源码分析，内容包括
- HDFS目录树管理
- HDFS元数据双缓存方案深度剖析
- HDFS元数据写JournalNode流程分析
- StandbyNameNode Checkpoint原理深度剖析
- ActiveNameNode FSimage文件更新流程分析
1. HDFS目录树管理
想研究 HDFS目录树管理,那么就写个创建目录的demoe，跟踪下代码看看具体实现，代码如下：
public class FIledemo {
public static void main(String[] args) throws IOException {
Configuration configured = new Configuration();
FileSystem fileSystem = FileSystem.newInstance(configured);
fileSystem.mkdirs(new Path("/data/hive/user/test"));
}
}
-
进入
fileSystem.mkdirs后，就一行代码跟进：
image -
发现是个抽象方法，所以需要找打它的实现。先看看类的注释 The local implementation is {@link LocalFileSystem}
and distributed implementation is DistributedFileSystem. 因为我们要研究的是分布式的，所有肯定是找DistributedFileSystem了
image -
进入
DistributedFileSystem然后找到mkdri()方法，发现又是一行代码，直接跟进
image -
找到
return dfs.mkdirs(getPathName(p), permission, createParent);然后跟进。这里是new了一个FileSystemLinkResolver对象，重写了该类的doCall和next方法后，后又调用FileSystemLinkResolver对象的resolve方法。我们进入该方法看看。
image -
可以看到这里实际上又调用了自己的
doCall方法，也就是刚才重写的那个第4步的doCall方法
image -
看着就只有最后一样的
return primitiveMkdir(src, masked, createParent);跟创建目录有关
image -
一眼就看到有个
namenode.mkdirs,这个太明显了，肯定就是它了。
image -
跟进一看又是一个抽象方法，在看实现类，有个
NameNodeRpcServer，我们记得NameNodeRpcServer就是Namenode的RPC服务，而管理元数据的操作都在NameNodeRpcServer，所以我们直接跟进去看看。
image
PS:这里留个疑问，为什么不进入ClientNamenodeProtocolTranslatorPB查看呢 -
在
NameNodeRpcServer中找到 mkdirs 方法，可以看到最后一行应该就是创建目录
image -
简单看看就知道肯定是
FSDirMkdirOp.mkdirs了
image -
仔细看代码后发现
createChildrenDirectories这个比较像
image -
跟进
createChildrenDirectories后，看注释说是沿着路径创建父/子目录和所有祖先目录。应就是了。而这里的核心代码肯定就是for循环里的existing = createSingleDirectory(fsd, existing, component, perm);
image -
这里也就看着
unprotectedMkdir像是创建目录，后面的代码应该是在创建完成后判断是否成功的，以及写editlog的。
image -
这里 new 了个
INodeDirectory对象。这对象就是用来存储目录信息的。
image -
我们跟进构造函数看看，就一样代码，是调用了父类的构造函数，继续跟进
image -
这里又是一行代码，调用了一个重载的构造函数
image -
再跟进，可以看到这里都是些赋值，包括
id，name,permission,modificationTime,accessTime,其中name就是目录的名字。
image -
返回
INodeDirectory可以看到里面还有一行private List<INode> children = null;代码，看到这个应该想到，INodeDirectory就应该是个构建目录树的类。一个INodeDirectory对象对应一个目录,可以看到INodeDirectory对象继承了INodeWithAdditionalFields，而INodeWithAdditionalFields又继承了INode,INode其实就是模仿了uninx的inode。
image -
我们继续14步的代码，在创建完
INodeDirectory后，后面会把它交给fsd.addLastINode处理，跟进去后，会发现把这个INodeadd 给parent
image -
进入 add 方法后看到又一个
addChild方法
image -
跟进
addChild方法后，又找到一个children.add，这里的children是个list集合，而且它就是我们第17步说的那个private List<INode> children = null;。
image
至此 HDFS 目录树的构建源码分析就结束了。我们来个图总结下：
2. HDFS元数据双缓存方案深度剖析
前面介绍了HDFS目录树的构建，但这只是在内存中的目录树，我们知道目录树在构建的时候还应写到磁盘上和Journalnode上。接下来我们在讲讲，目录树如何刷新到磁盘和Journalnode吧。
-
在HDFS目录树管理的第13步，我们看到了一个关于Editlog的代码
fsd.getEditLog().logMkDir(cur, newNode);，这个代码我们跟进下看看：
image -
可以看到这里是构建了一个
MkdirOp对象，然后又把这个对象传入了logEdit方法
image -
在
logEdit里，可以看到先beginTransaction()，这个是把txid变量自增以下，然后又把MkdirOp对象写到一个流中editLogStream.write(op)。最后调用了logSync();方法。这里的editLogStream我们在下一节会详细介绍。
image logSync();方法的注释是：将缓冲编辑日志项同步到永久存储，此方法非常重要，这里就是 Namenode 管理元数据的双缓冲和分段加锁机制的实现。我们先看注释
Sync all modifications done by this thread.
The internal concurrency design of this class is as follows:
- Log items are written synchronized into an in-memory buffer,
and each assigned a transaction ID.
- When a thread (client) would like to sync all of its edits, logSync()
uses a ThreadLocal transaction ID to determine what edit number must
be synced to.
- The isSyncRunning volatile boolean tracks whether a sync is currently
under progress.
The data is double-buffered within each edit log implementation so that
in-memory writing can occur in parallel with the on-disk writing.
Each sync occurs in three steps:
1. synchronized, it swaps the double buffer and sets the isSyncRunning
flag.
2. unsynchronized, it flushes the data to storage
3. synchronized, it resets the flag and notifies anyone waiting on the
sync.
The lack of synchronization on step 2 allows other threads to continue
to write into the memory buffer while the sync is in progress.
Because this step is unsynchronized, actions that need to avoid
concurrency with sync() should be synchronized and also call
waitForSyncToFinish() before assuming they are running alone.
注释太长我就一句句翻译了，大概的意思就是有两个内存缓冲（内存1和内存2），内存1来接受日志，内存2用来向磁盘或Journal写日志。具体写的过程分三步走：
- 达到刷新磁盘的条件后进行内存交换 - 加同步锁
- 将内存中的数据刷新到磁盘 - 不加同步锁
- 重置标记并通知等待其他线程 - 加同步锁
由于 logSync() 太长，里面的代码有都很重要，所以我按照三个步骤分别截图讲解。
4.1 首先定义了2个变量，并获得线程事务的ID，下面进入同步代码块
- 在同步代码块中，先判断是否已经有线程已经在同步了（如果有就等待1秒），
下面又判断该线程的日志是否已经被刷新到内存2中，如果是直接return了。 - 然后继续就变更了这三个变量
syncStart = txid;isSyncRunning = true;sync = true;。这三个表syncStart代表内存2中最后一个日志的编号（注意所有编号是有序自增的，具体可以看这个方法beginTransaction()），isSyncRunning代表当前是否有线程在同步。sync具体意义未知 - 然后就交换内存1和内存2的指针。
4.2. 这一步比较简单就是对内存2进行写磁盘操作，具体的实现可以看EditLogFileOutputStream 和 JournalSetOutputStream
4.3. 第三步也比较简单就是更改一些标记，并唤醒其他线程
值得注意的地方是，在HDFS目录树管理第9步也会触发 editlog 日志向磁盘刷新的操作，实际调用的代码是一个地方。
至此双缓冲方案也分析完成了。画个图总结下：
3. HDFS元数据写JournalNode流程分析
在 HDFS元数据双缓存方案深度剖析 的第3步有个editLogStream变量，其实这个变量是有多个类型的，对应的不同功能，我们这里来说说
- 我们来看看这个
editLogStream的赋值，通过搜索可以找到editLogStream的赋值代码
image - 跟进后发现，这里显示调用了
mapJournalsAndReportErrors方法，然后又返回了一个JournalSetOutputStream对象。这样我们就在知道，这里是editLogStream就是个JournalSetOutputStream对象。而调用了editLogStream的write方法，就是调用了JournalSetOutputStream对象的write方法。
image - 进入
JournalSetOutputStream对象的write方法后，可以看到又有一个write()方法,但是这个方法是抽象方法，必须找到它的实现类。通过查询，我们发现这个实现类有 4 个。具体是哪个呢，我们先看看mapJournalsAndReportErrors方法。
image - 进入
mapJournalsAndReportErrors,有注释就先看注释。通过注释大概了解到应该是调用JournalSet.JournalClosure中的操作。看具体的代码，感觉也就for循环和这两行有点用。
image - 我们先看看for循环中的
journals里是什么。搜索journals.add 发现有如下代码。我们在看看具体是哪里调用了这个add(JournalManager j, boolean required, boolean shared)方法
image -
结果发现有3个地方调用了，我们分别根据看看
image - 第1个和第2个都是在
FSEditLog的一个地方调用了，第1个这里add的是FileJournalManager对象，因为前面有个判断u.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)，意思是判断该URI是否本地文件。
image - 第2个add的是
QuorumJournalManager,这里使用的是反射的方式,具体的class路径在配置文件中
image
image
image - 第3个是被自己调用，然后又跟进的话,发现也是被
FSEditLog调用了，这里add的是BackupJournalManager对象。这个一看就是备份用的，跟我们现在查看的应没什么关系，先pass。
image
image - 从第5步到第9步，说了那么多其实就是说明，第3步的for循环的列表里的
JournalAndStream对象里JournalManager其实是FileJournalManager或QuorumJournalManager。我们接着看第3步看，getCurrentStream这里返回了一个EditLogOutputStream对象，那么这个对象在哪里赋值的呢？
image - 发现是在这里赋值的，而
journal变量就是我们之前说的FileJournalManager或QuorumJournalManager。
image - 我们先看看
FileJournalManager，它返回的是EditLogFileOutputStream对象。
image - 我们再看看
QuorumJournalManager，他返回的是QuorumOutputStream对象。
image - 至此我们了解到第3步那，是调用了
EditLogFileOutputStream或QuorumOutputStream的write方法。我们分别看看。 -
EditLogFileOutputStream的write方法
image - 再跟下去，就发现了Editlog双缓冲用到的
EditsDoubleBuffer类。而write方法就是往这个类里提供的TxnBuffer bufCurrent变量里写入数据。
image
EditLogFileOutputStream类里还有两个方法setReadyToFlush和flushAndSync,分别对应了EditsDoubleBuffer里的setReadyToFlush和flushTo,分别是内存交换和刷新数据到磁盘，实现都比较简单了。具体的可以自己看看吧。
image -
QuorumOutputStream的write方法
image -
QuorumOutputStream的write和setReadyToFlush也是使用的EditsDoubleBuffer类的实现。但是flushAndSync方法就不是了（这里是写journalnode集群） - 这里看看
QuorumOutputStream的flushAndSync。可以看到，这里显示把原来缓存的数据复制到另一个流中（这里主要是为了安全），然后又使用sendEdits方法把数据发送出去。
image - 进入
sendEdits，可以看到有个for循环，这里就是遍历所有的AsyncLogger,而每个AsyncLogger其实就对应一个 journalnode 节点。所有的AsyncLogger都调用了sendEdits而它是个抽象方法，它就一个实现，直接看实现就行。
image - 在
sendEdits的实现类里，发现了获得了一个代理然后调用代理的journal方法
image - 这个代理就是
QJournalProtocol，而它的实现就是JournalNodeRpcServer。我们直接看JournalNodeRpcServer的journal方法吧。
image - 就一行代码，然后调用了
journal方法，再跟进
image - 跟进
journal后，看到有个writeRaw方法，这里就是写数据的地方了。
image
终于把这块跟踪完了，最后简答画个图：
4. StandbyNameNode Checkpoint原理深度剖析
StandbyNameNode 同步 editlog日志，主要是通过EditLogTailer类。先看类的注释。
EditLogTailer represents a thread which periodically reads from edits journals
and applies the transactions contained within to a given FSNamesystem.
EditLogTailer表示一个线程，它定期从edits日志中读取数据，
并将其中包含的事务应用于给定的FSNamesystem。
既然EditLogTailer是个线程，那么它肯定有run方法。
- 找到
EditLogTailer的run方法。就一行代码doTailEdits()
image - 跟进
doTailEdits，可以方法这里显示获得了一个FSImage对象，然后又调用了它的loadEdits方法。
image - 跟进
loadEdits后，发现里面就一行代码继续跟进。
image - 又是获得了一个
FSEditLogLoader，然后调用了它的loadFSEdits方法。
image - 进入
loadFSEdits后，看到一个注释，简单看下
Load an edit log, and apply the changes to the in-memory structure
This is where we apply edits that we've been writing to disk all along.
此方法里看着loadEditRecords像是加载日志。我们进入看看
- 进入
loadEditRecords后，简单过下代码，会看到一个while(true)，里面还有一个in.readOp()这个看着就像读取操作。而且in.readOp()返回的是个FSEditLogOp对象。而FSEditLogOp其实就是编辑日志所对应的对象。具体怎么读取的我们先不管，我们先看看下面。
image - 在往后看，可以看到
applyEditLogOp方法,我们进这个方法看看
image - 进到
applyEditLogOp方法后，发现里面是个switch，我们本文最开始的场景是创建目录，所以我们直接找到mkdir的case看看。发现有一个mkdirForEditLog方法。
image - 进入
mkdirForEditLog看看，结果发现了一个unprotectedMkdir方法。还记得 HDFS目录树管理 里的讲解吗。里面就提到这个方法。
image - 我们跟进看看，这里就是我们在 HDFS目录树管理 里提到的代码。还记得吧，这里是创建了一个
INodeDirectory对象，然后把INodeDirectory添加到另外的INodeDirectory对象的child里。最后形成一个目录树。详细的这里就不跟了。
image - 以上就是
StandBy状态的Namenode同步Editlog的过程。不过上面并没有说明如何读取的Journal。这里简单说说吧，我们记得第6步中有个in.readOp()方法吧。跟进去看看,然后发现其实就是执行了nextOp()方法。
image - 进入
nextOp()，就一行代码。
image - 在进入
nextOpImpl()方法，可以看到一个switch。具体的步骤是根据state的值进行选择的。我们看看state的值什么。
image -
state默认就是UNINIT,那我们case UNINIT的代码。
image -
case UNINIT里调用了一个init的代码。可以看到这里初始化堆变量后，又把state设置为OPEN了。init代码执行完后，又调用了nextOpImpl(),所以，这次就到了case OPEN了。
image - 可以看到
case OPEN的代码里，主要的就是op = reader.readOp(skipBrokenEdits);。我们现在要搞明白这个reader是什么。
image - 搜索
reader的赋值，结果我们又回到了init方法。原来case UNINIT就是给reader赋值的。那么我看仔细看，会发现reader最终使用的是这个log变量的输入流（这里使用了装饰模式）。那么我们还是要确定log变量是什么。
image - 我们搜索下，发现
log变量是在构造方法中赋值的。 我们看下是哪里调用了这个构造方法
image - 接着发现有两个地方调用了，其中一个传的值是
FileLog另一个是UrlLog。因为我们是要从Journal上同步数据，FileLog一看就像是本地读取的。而UrlLog应该更符合我们的需求。所以这个log对象更应该是UrlLog对象。那么在第17步里调用了log.getInputStream()那实际上应该使用的就是UrlLog的getInputStream。
image - 这里可以看到实际上返回的是一个
HttpURLConnection的链接。那么既然是http链接，而又是应用给journal的，那是不是journal提供了一个http服务呢？我们马上验证下。
image - 我们简单搜索后发现，果然有个类叫
JournalNodeHttpServer。里面有个start方法，构建了一个httpserver2服务，并且该服务只提供了一个servlet，路径就是/getJournal。既然是servlet那么肯定就有doGet或doPost方法。我们看看。
image - 进入
doGet后发现，原来这里有个一个流对拷。
image
最后画个图简单总结下
image
思考为什么已经提供RPC服务了，为什么又提供一个HTTP服务呢？
5. ActiveNameNode FSimage文件更新流程分析
NameNode 做FSimage文件更新的类是StandbyCheckpointer。 老样子先看类注释：
/**
* Thread which runs inside the NN when it's in Standby state,
* periodically waking up to take a checkpoint of the namespace.
* When it takes a checkpoint, it saves it to its local
* storage and then uploads it to the remote NameNode.
*/
- 它是个线程，并运行在 Standby 状态的 NameNode 上。
- 周期性的获取 namespace 的 checkpoint。
- 当获得checkpoint时，会保存到本地并更新它到Active的 NameNode。
- 既然是是个线程，那么我们直接看他的
run()方法。
image
废话不说，直接进入doWork()方法。 - 进入
doWork()后，会发现代码很长。我们分段来看
- 2.1 一看就是个
while循环，开始就是个线程等待。没什么可说的继续下面。
image - 2.2
1是判断登录的与我们的场景关系不大，2是检测是否要回滚与我们的场景关系不大，3是检查未 checkpoint 的日志数量，4是检查上次 checkpoint 的时间的。第3和第4会执行needCheckpoint = true;代码。 动作。具体条件是如果未 checkpoint 的日志量超 100万 或上次 checkpoint 时间超过 1小时。这里只是设置一个 checkpoint 的标记。我们继续看下面代码。
image -
2.3 这里是个故障判断，与我们的场景关系不大，过了。
image - 2.4
doCheckpoint();就是开始 checkpoint 了,进入doCheckpoint();。
image
-
doCheckpoint();代码又很长，还是分段看。
- 3.1
1那就是定义几个变量，2那就是获得一个锁，3判断当前editlog的最大id和将要更新的editlog的最大id是否一致，其实就是判断是否已经更新过了。
image - 3.2
1判断是不是回滚的2的img.saveNamespace(namesystem, imageType, canceler);就是 执行checkpoint3判断checkpoint是否成功的,跟进img.saveNamespace(namesystem, imageType, canceler);看看
image
- 这里就进入
FSImage类了。1就是检查editlog是否是写的状态；2就是获得editlog的最新日志id;3这里开始进行checkpoint，这个里面就是把内存的 editlog写入磁盘文件中了。具体就先不跟进了
image - 回到
StandbyCheckpointer我们接着第3步，看看磁盘存储后，进行的步骤。看1的注释：上传checkpoint到active Namenode;2这里就是实际的代码了。
image - 跟进
TransferFsImage.uploadImageFromStorage方法。1：注意这里的URL是使用了的ImageServlet.PATH_SPEC,也就是说等会我们就要看ImageServlet的代码了。2这里就是请求url了。我们先看看2的方法TransferFsImage.uploadImage
image - 进入
TransferFsImage.uploadImage,这里的代码又比较多，分开看看。
- 7.1
1就是获得 Standby 状态Namenode上的FSImage文件。2构建一个URIBuilder对象，3就是个给URIBuilder对象传参呗。这里都很简单。
image - 7.2
1就是构建HttpURLConnection对象，下面就是给这个对象设置参数，值得注意的地方是这里设置的是put请求；2就是真正的请求url写文件了。进入TransferFsImage.writeFileToPutRequest。
image
-
TransferFsImage.writeFileToPutRequest，这里1:还是给HttpURLConnection对象设置参数；2是获得输入和输出流；3就是开始流的拷贝了。进入TransferFsImage.copyFileToStream。
image -
TransferFsImage.copyFileToStream，到这里就是IO操作了，没啥可说的。我们去看看ImageServlet.doput方法那接收到做了什么操作吧。
image -
ImageServlet.doput,之类代码比较长，分开看看。
- 10.1 这里也没啥，就是获得
request的参数，然后做了些校验之类的。
image - 10.2 这里也很简单了，
1就是获得输入流;2就是对数据进行小校验;3就是真正的替换本地的FSImage文件了。后面还有好几步，但都很简单最终就是用file.write()写到本地就完了。
image
老样子，用一个张图简单总结下
image
