学步园

继续为分析org.apache.hadoop.hdfs.server.namenode.FSNamesystem类做准备，这里分析与FSEditLog相关的几个类，当然，FSEditLog类才是核心的。

• FSEditLog.EditLogFileOutputStream内部静态类

该类是定义在org.apache.hadoop.hdfs.server.namenode.FSEditLog类内部的静态类，表示将EditLog日志通过打开一个该输出流实例写入到本地磁盘。该输出流类的继承层次结构如下所示：

◦java.io.OutputStream(implements java.io.Closeable, java.io.Flushable)<br /> ◦org.apache.hadoop.hdfs.server.namenode.EditLogOutputStream<br /> ◦org.apache.hadoop.hdfs.server.namenode.FSEditLog.EditLogFileOutputStream

首先，看EditLogOutputStream这个抽象类的定义。

EditLogOutputStream类是一个用来支持将EditLog日志文件中内容持久化（写入）到存储目录中的通用抽象类。该抽象类定义了两个统计变量，如下所示：

private long numSync; // 对EditLog日志文件执行同步操作（同步写入磁盘）的次数<br /> private long totalTimeSync; // 同步EditLog日志文件到磁盘，所花费的总累加时间

该抽象类中实现的方法都是与更新统计变量相关的，如下所示：

/**<br /> * 将数据持久化到存储目录，保持数据同步一致，同时刷新统计数据<br /> */<br /> public void flush() throws IOException {<br /> numSync++; // 执行该方法一次，则同步数据统计变量加1<br /> long start = FSNamesystem.now(); // 此次开始同步的时刻（用来计算此次同步所花费的时间，记入对应统计变量中）<br /> flushAndSync(); // 调用刷新与执行同步的操作<br /> long end = FSNamesystem.now(); // 此次同步结束时间<br /> totalTimeSync += (end - start); // 计算此次同步花费时间，累加到统计变量totalTimeSync上<br /> }</p> <p> /**<br /> * 获取执行flushAndSync()同步方法花费的总时间<br /> */<br /> long getTotalSyncTime() {<br /> return totalTimeSync;<br /> }</p> <p> /**<br /> * 获取调用flushAndSync()方法执行同步的次数<br /> */<br /> long getNumSync() {<br /> return numSync;<br /> } 

该抽象类中定义了一些抽象方法，这些方法需要在该类的子类中给出具体实现：

/**<br /> * 获取该EditLogOutputStream输出流的名称<br /> */<br /> abstract String getName();</p> <p> /**<br /> * 该方法继承自OutputStream，表示向该输出流对象中写入字节<br /> */<br /> abstract public void write(int b) throws IOException;</p> <p> /**<br /> * 将EditLog日志记录写入到该输出流对象中<br /> * （日志记录通过操作名称和一个Writable参数的数组来表征）<br /> */<br /> abstract void write(byte op, Writable ... writables) throws IOException;</p> <p> /**<br /> * 创建并初始化一个新的EditLog日志存储对象<br /> */<br /> abstract void create() throws IOException;</p> <p> /** {@inheritDoc} */<br /> abstract public void close() throws IOException;</p> <p> /**<br /> * 所有已经被写入到该输出流对象中的数据，将准备对其执行刷新（flush）操作。<br /> * （在执行刷新的过程中，新的数据仍然能够被写入到该输出流对象中）<br /> */<br /> abstract void setReadyToFlush() throws IOException;</p> <p> /**<br /> * Flush并且sync所有准备好的数据（调用了setReadyToFlush()方法表示准备），持久化到相应的（持久）存储中<br /> */<br /> abstract protected void flushAndSync() throws IOException;</p> <p> /**<br /> * 获取当前EditLog日志文件的大小<br /> * 得到EditLog日志文件的长度是为了检查日志文件是否大到需要启动一个检查点进程来处理<br /> */<br /> abstract long length() throws IOException;

通过对EditLogOutputStream抽象类的分析，我们获知：一个EditLog日志文件也是通过流式进行写的，而且定时对日志文件进行同步，写入持久存储中，其中支持写入字节和Writable类型数据；在同步的过程中，需要记录同步统计数据；当EditLog日志文件达到一定大小的时候，需要启动检查点进程来进行处理。

然后，我们看FSEditLog.EditLogFileOutputStream内部静态类的具体实现。

该类继承自EditLogOutputStream抽象类，实现了该抽象类中定义的抽象方法，能够在一个本地文件（local file）中存储EditLog日志文件。

该类定义了如下属性：

private File file; // EditLog日志文件记录所存储的本地文件<br /> private FileOutputStream fp; // 存储EditLog日志记录的文件输出流对象<br /> private FileChannel fc; // 执行同步操作（sync）的文件输出流对象对应的通道<br /> private DataOutputBuffer bufCurrent; // 当前用于写操作的数据缓冲区<br /> private DataOutputBuffer bufReady; // 为了执行写操作而准备的数据缓冲区<br /> static ByteBuffer fill = ByteBuffer.allocateDirect(512); // 预分配大小为512的字节缓冲区

下面是FSEditLog.EditLogFileOutputStream类的构造方法：

EditLogFileOutputStream(File name) throws IOException {<br /> super();<br /> file = name; // EditLog日志文件对应的本地文件名称<br /> bufCurrent = new DataOutputBuffer(sizeFlushBuffer); // 分配sizeFlushBuffer = 512*1024<br /> bufReady = new DataOutputBuffer(sizeFlushBuffer); // 分配sizeFlushBuffer = 512*1024<br /> RandomAccessFile rp = new RandomAccessFile(name, "rw"); // 根据文件File name创建一个随机读写文件实例<br /> fp = new FileOutputStream(rp.getFD()); // 为向fp输出流对象中追加数据而打开该输出流<br /> fc = rp.getChannel(); // 获取fp对应的通道<br /> fc.position(fc.size()); // 设置输出流通道对应文件流的当前位置<br /> }

通过上面的构造方法，可见通过构造方法已经准备好了写EditLog日志文件的条件，定位到EditLog对应于本地存储文件流的当前写入位置，也就是说对日志文件的操作时追加写操作。

下面看该类中实现的方法：

@Override<br /> String getName() {<br /> return file.getPath(); // 返回EditLog日志文件对应的本地存储文件的路径<br /> }</p> <p> @Override<br /> public void write(int b) throws IOException {<br /> bufCurrent.write(b); // 将字节写入当前数据缓冲区bufCurrent<br /> }</p> <p> @Override<br /> void write(byte op, Writable ... writables) throws IOException {<br /> write(op); // 调用重载的write方法，写入操作名称（其实是一个操作代码）<br /> for(Writable w : writables) { // 迭代Writable对象数组<br /> w.write(bufCurrent); // 分别写入到前数据缓冲区bufCurrent<br /> }<br /> }</p> <p> /**<br /> * 创建一个空的EditLog日志文件<br /> */<br /> @Override<br /> void create() throws IOException {<br /> fc.truncate(0); // 设置当前文件的大小（如果文件大小大于0，删除文件中的全部字节数据）<br /> fc.position(0); // 设置当前写位置（初始化）<br /> bufCurrent.writeInt(FSConstants.LAYOUT_VERSION); // 在EditLog日志文件最前面写入版本号<br /> setReadyToFlush(); // 设置准备刷新状态<br /> flush(); // 执行刷新操作（将版本号写入到输出流中）<br /> }</p> <p> @Override<br /> public void close() throws IOException {<br /> // 在所有追加写事务执行flush与sync操作完成以后，该方法被调用<br /> int bufSize = bufCurrent.size();<br /> if (bufSize != 0) {<br /> throw new IOException("FSEditStream has " + bufSize + " bytes still to be flushed and cannot " + "be closed.");<br /> }<br /> bufCurrent.close(); // 关闭当前数据缓冲区<br /> bufReady.close(); // 关闭预执行flush准备数据缓冲区</p> <p> fc.truncate(fc.position()); // 从事务日志中删除最后面的OP_INVALID标记（因为在调用setReadyToFlush方法结束时，写入了OP_INVALID标记）<br /> fp.close(); // 关闭EditLog日志文件对应的本地存储文件输出流对象</p> <p> bufCurrent = bufReady = null; // 释放<br /> }</p> <p> /**<br /> * 所有已经被写入到该输出流对象中的数据，将准备对其执行刷新（flush）操作。<br /> * （在执行刷新操作的过程中，新的数据仍然能够被写入到该输出流对象中）<br /> */<br /> @Override<br /> void setReadyToFlush() throws IOException {<br /> assert bufReady.size() == 0 : "previous data is not flushed yet";<br /> write(OP_INVALID); // 在文件末尾写入一个标记OP_INVALID<br /> DataOutputBuffer tmp = bufReady; // 切换缓冲区：将当前的bufCurrent切换为准备好要执行flush操作的bufReady<br /> bufReady = bufCurrent;<br /> bufCurrent = tmp;<br /> }</p> <p> /**<br /> * 将bufCurrent中数据到持久存储中<br /> * currentBuffer is not flushed as it accumulates new log records<br /> * while readyBuffer will be flushed and synced.<br /> */<br /> @Override<br /> protected void flushAndSync() throws IOException {<br /> preallocate(); // 如果必要的话，为文件预分配内存<br /> bufReady.writeTo(fp); // 将bufReady中数据写入到文件<br /> bufReady.reset(); // 重启bufReady数据缓冲区<br /> fc.force(false); // 因为调用preallocate()方法对文件进行了预分配，也就没必要更新该文件元数据信息<br /> fc.position(fc.position()-1); // 设置当前写入位置：回退排除文件末尾的标记OP_INVALID<br /> }</p> <p> /**<br /> * 获取当前EditLog日志文件的大小（包括bufReady与bufCurrent）<br /> */<br /> @Override<br /> long length() throws IOException {<br /> return fc.size() + bufReady.size() + bufCurrent.size(); // 返回文件大小 = 文件实际大小 + bufReady大小 + bufCurrent大小<br /> }</p> <p> // 为日志文件预分配一个大数据块<br /> private void preallocate() throws IOException {<br /> long position = fc.position(); // 获取文件当前位置<br /> if (position + 4096 >= fc.size()) {<br /> FSNamesystem.LOG.debug("Preallocating Edit log, current size " + fc.size());<br /> long newsize = position + 1024*1024; // 1MB<br /> fill.position(0); // 预分配数据缓冲区fill写入位置为0<br /> int written = fc.write(fill, newsize); // 为日志文件增加分配newsize个字节<br /> FSNamesystem.LOG.debug("Edit log size is now " + fc.size() + " written " + written + " bytes " + " at offset " + newsize);<br /> }<br /> }</p> <p> /**<br /> * 获取与该输出流相关联的文件<br /> */<br /> File getFile() {<br /> return file;<br /> }

通过对上面方法的阅读分析，可以了解到，EditLog日志文件是与本地磁盘上的一个文件相对应的，实际也是写入到这个本地文件中的。在执行流式追加写的过程中，设置了两个重要的数据缓冲区，bufReady 与bufCurrent，通过切换这两个数据缓冲区来将不断追加写入到EditLogFileOutputStream流对象中的数据，同步到本地磁盘文件中。而且，当执行flushAndSync方法进行同步时候，如果必要会对EditLog日志文件对应的本地文件进行预分配，保证可持续做好写事务操作的记录。

• FSEditLog.EditLogFileInputStream内部静态类

该类org.apache.hadoop.hdfs.server.namenode.FSEditLog.EditLogFileInputStream是对EditLog日志文件进行读取的输入流实现类。该输入流类与EditLogFileOutputStream输出流类的组织关系类似，它的继承层次关系如下所示：

◦java.io.InputStream(implements java.io.Closeable)<br /> ◦org.apache.hadoop.hdfs.server.namenode.EditLogInputStream<br /> ◦org.apache.hadoop.hdfs.server.namenode.FSEditLog.EditLogFileInputStream

首先，看抽象流类EditLogInputStream。该类是一个用来支持读取本地存储目录中对应的文件，从而得到EditLog日志文件的通用抽象类。该抽象类比较简单，定义获取继承了一组抽象方法，如下所示：

/**<br /> * 获取该输入流的名称<br /> */<br /> abstract String getName();</p> <p> /**<br /> * 估算大概存在多少字节可读的数据<br /> */<br /> public abstract int available() throws IOException;</p> <p> /**<br /> * 从该输入流中读取下一个字节的数据<br /> */<br /> public abstract int read() throws IOException;</p> <p> /**<br /> * 从该输入流off位置读取len个字节到字节数组b中<br /> */<br /> public abstract int read(byte[] b, int off, int len) throws IOException;</p> <p> /**<br /> * 关闭该输入流，释放相关资源<br /> */<br /> public abstract void close() throws IOException;</p> <p> /**<br /> * 获取当前EditLog日志文件的大小<br /> */<br /> abstract long length() throws IOException;

没什么可说的，继续看FSEditLog.EditLogFileInputStream内部静态类的具体实现吧。

FSEditLog.EditLogFileInputStream内部静态类也比较简单，就是提供了对EditLog日志文件的读取操作，也是基于流的，该类源代码如下所示：

static class EditLogFileInputStream extends EditLogInputStream {<br /> private File file; // EditLog日志文件对应的本地文件<br /> private FileInputStream fStream; // 文件file对应的输入流对象</p> <p> EditLogFileInputStream(File name) throws IOException { // 构造方法<br /> file = name;<br /> fStream = new FileInputStream(name); // 构造file的输入流实例<br /> }</p> <p> @Override<br /> String getName() { // 获取file的路径<br /> return file.getPath();<br /> }</p> <p> @Override<br /> public int available() throws IOException {<br /> return fStream.available(); // 获取从该输入流可以读取到的字节数<br /> }</p> <p> @Override<br /> public int read() throws IOException {<br /> return fStream.read(); // 从该输入流中读取一个字节的数据<br /> }</p> <p> @Override<br /> public int read(byte[] b, int off, int len) throws IOException {<br /> return fStream.read(b, off, len); // 将该输入流中从off位置开始读取len个字节到字节数组b中<br /> }</p> <p> @Override<br /> public void close() throws IOException {<br /> fStream.close(); // 关闭该输入流<br /> } </p> <p> @Override<br /> long length() throws IOException {<br /> return file.length(); // 返回文件大小 = 文件实际大小 + bufReady大小 + bufCurrent大小<br /> }<br /> }

• FSEditLog类

该类维护一个记录对文件系统命名空间进行修改操作的日志文件。

我们已经对FSEditLog.EditLogFileOutputStream类与FSEditLog.EditLogFileInputStream类非常熟悉了，它们是与EditLog日志文件的读写密切相关的。

在FSEditLog类内部还有一个实现了Writable接口的可序列化内实体类FSEditLog.BlockTwo，它能够从以旧格式存储的块中读写数据，如下所示：

static class BlockTwo implements Writable {<br /> long blkid; // 块ID<br /> long len; // 块长度</p> <p> static { // 注册<br /> WritableFactories.setFactory<br /> (BlockTwo.class,<br /> new WritableFactory() {<br /> public Writable newInstance() { return new BlockTwo(); }<br /> });<br /> }</p> <p> BlockTwo() {<br /> blkid = 0;<br /> len = 0;<br /> }</p> <p> public void write(DataOutput out) throws IOException {<br /> out.writeLong(blkid);<br /> out.writeLong(len);<br /> }</p> <p> public void readFields(DataInput in) throws IOException {<br /> this.blkid = in.readLong();<br /> this.len = in.readLong();<br /> }<br /> }

通过一个BlockTwo对象，能够读写块的ID和块的长度。 

另一个是事务ID的封装内部实体类FSEditLog.TransactionId，表示EditLog日志文件记录的修改操作这样的事务ID，在FSEditLog类中主要是用来作为当前执行事务的线程对应的线程局部变量的拷贝：ThreadLocal<TransactionId>。该类如下所示：

private static class TransactionId {<br /> public long txid;</p> <p> TransactionId(long value) {<br /> this.txid = value;<br /> }<br /> }

FSEditLog.TransactionId类只封装了一个事务ID，再没有其它内容了。 

我们再看FSEditLog类。

首先看，对文件系统命名空间进行不同操作的操作代码，通过向EditLog日志文件中写入对应的操作代码来表示实际的操作，这些操作代码包含：

private static final byte OP_INVALID = -1; // 非法操作<br /> private static final byte OP_ADD = 0; // 添加文件操作<br /> private static final byte OP_RENAME = 1; // 重命名<br /> private static final byte OP_DELETE = 2; // 删除<br /> private static final byte OP_MKDIR = 3; // 创建目录<br /> private static final byte OP_SET_REPLICATION = 4; // 设置副本因子<br /> // 下面两个仅仅为了保证后向兼容<br /> @Deprecated private static final byte OP_DATANODE_ADD = 5;<br /> @Deprecated private static final byte OP_DATANODE_REMOVE = 6;<br /> private static final byte OP_SET_PERMISSIONS = 7; // 设置权限<br /> private static final byte OP_SET_OWNER = 8; // 设置属主<br /> private static final byte OP_CLOSE = 9; // 写操作完成后的关闭操作<br /> private static final byte OP_SET_GENSTAMP = 10; // 存储时间戳<br /> // 下面两个已经不再使用，如果LAST_UPGRADABLE_LAYOUT_VERSION等于-17或者更新，则可以删除掉<br /> private static final byte OP_SET_NS_QUOTA = 11; // 设置文件系统命名空间配额<br /> private static final byte OP_CLEAR_NS_QUOTA = 12; // 清除文件系统命名空间配额<br /> private static final byte OP_TIMES = 13; // 为一个文件设置mod & access时间<br /> private static final byte OP_SET_QUOTA = 14; // 设置名称和磁盘配额

再看该类中定义的其它属性，如下所示：

private static int sizeFlushBuffer = 512*1024; // 数据缓冲区大小，主要是写日志文件过程中为两个执行切换的数据缓冲区分配大小<br /> private ArrayList<EditLogOutputStream> editStreams = null; // EditLogOutputStream输出流对象列表<br /> private FSImage fsimage = null; // FsImage映像</p> <p> private long txid = 0; // 单调递增事务ID计数器，用来为事务分配ID的<br /> private long synctxid = 0; // 最后执行sync同步操作的事务ID<br /> private long lastPrintTime; // 最后一次将统计数据输出到日志文件的时间<br /> private boolean isSyncRunning; // 是否当前正在执行sync同步操作</p> <p> private long numTransactions; // 事务数量统计变量<br /> private long numTransactionsBatchedInSync; // 批量sync事务的数量的统计变量<br /> private long totalTimeTransactions; // 执行全部事务的时间统计变量<br /> private NameNodeMetrics metrics; // Namenode统计数据的统计变量</p> <p> // 存储线程的当前事务<br /> private static final ThreadLocal<TransactionId> myTransactionId = new ThreadLocal<TransactionId>() {<br /> protected synchronized TransactionId initialValue() {<br /> return new TransactionId(Long.MAX_VALUE);<br /> }<br /> };

该类的构造方法如下所示：

FSEditLog(FSImage image) { // 通过FSImage实例来构造，FSImage类会在之后分析<br /> fsimage = image;<br /> isSyncRunning = false; // 当前没有进行同步sync<br /> metrics = NameNode.getNameNodeMetrics(); // 通过Namenode获取到其状态统计数据对象，用来跟踪与EditLog相关的信息（写入Namenode的统计数据对象中）<br /> lastPrintTime = FSNamesystem.now(); // 初始化时间<br /> } 

关于一个FSEditLog类的实例能够做哪些事情，我们从该类中挑选几个比较关键的方法来详细解释说明。

1、创建一个EditLog日志文件

如下所示：

public synchronized void createEditLogFile(File name) throws IOException {<br /> EditLogOutputStream eStream = new EditLogFileOutputStream(name); // 根据指定的文件名，构造EditLog日志文件的输出流<br /> eStream.create(); // 创建EditLog日志文件，并写入版本号LAYOUT_VERSION，同时设置当前待写入的位置<br /> eStream.close(); // 关闭输出流<br /> }

另外，还有一个用来表示当前最新的edit文件edit.new，当该文件丢失的时候，需要进行创建，该操作对应于方法createNewIfMissing，如下所示：

synchronized void createNewIfMissing() throws IOException {<br /> for (Iterator<StorageDirectory> it = fsimage.dirIterator(NameNodeDirType.EDITS); it.hasNext();) { // 检查是否edit.new文件丢失<br /> File newFile = getEditNewFile(it.next());<br /> if (!newFile.exists())<br /> createEditLogFile(newFile); // 如果不存在，就调用上面的createEditLogFile创建一个<br /> }<br /> }  

2、打开 一个EditLog日志文件

方法open实现了打开日志文件的功能。实际上，当创建一个FSEditLog类的实例以后（通过构造方法构造得到），调用open方法打开一个该文件的输出流，准备后继向流中持续追加日志记录。open方法如下所示：

public synchronized void open() throws IOException {<br /> numTransactions = totalTimeTransactions = numTransactionsBatchedInSync = 0; // 初始化这些统计变量<br /> if (editStreams == null)<br /> editStreams = new ArrayList<EditLogOutputStream>(); // 初始化该日志文件实例的EditLogOutputStream流对象列表<br /> // 通过FSImage实例获取到Namenode对应的EDITS类型的存储目录的迭代器实例（因为这是一组存储目录，势必对应用来写EditLog日志的输出流EditLogFileOutputStream列表）<br /> // 注意：FSImage实例实际上只拥有一张登记了全部EDITS类型的存储目录的表，至于该存储目录中是否存在对应的文件，在加载的时候回检测出来，并采取相应的策略<br /> for (Iterator<StorageDirectory> it = fsimage.dirIterator(NameNodeDirType.EDITS); it.hasNext();) {<br /> StorageDirectory sd = it.next();<br /> File eFile = getEditFile(sd); // 获取一个EditLog文件<br /> try {<br /> EditLogOutputStream eStream = new EditLogFileOutputStream(eFile); // 打开该EditLog文件输出流<br /> editStreams.add(eStream); // 加入列表<br /> } catch (IOException e) {<br /> FSNamesystem.LOG.warn("Unable to open edit log file " + eFile);<br /> it.remove(); // 如果处理该目录sd发生异常，从存储目录列表中删除该目录sd<br /> }<br /> }<br /> }

调用open方法，其实是从Namenode来加载EditLog日志文件所对应的全部存储目录，并打开每一个与EditLog日志文件相关的本地文件的文件输出流，为了便于FSEditLog类实例管理这些输出流，使用一个ArrayList来存放于内存中。

3、处理输出流IO异常

通过前面分析，一个FSEditLog类的实例维护一个EditLogOutputStream输出流对象列表，那么当某个输出流对象发生IO异常的时候，需要对其作出处理，而不能影响其它事务的执行。 该类中给出了三个处理IO异常的方法，如下所示：

/**<br /> * 如果发生与EditLog日志相关的IO异常，会从editStreams列表中删除掉发生IO异常的输出流对象<br /> * 第一个：根据输出流对象在editStreams列表中索引来处理<br /> */<br /> synchronized void processIOError(int index);</p> <p> /**<br /> * 如果发生与EditLog日志相关的IO异常，会从editStreams列表中删除掉发生IO异常的输出流对象<br /> * 第二个：根据存储目录名称来处理editStreams列表中对应的输出流对象<br /> */<br /> synchronized void processIOError(StorageDirectory sd);</p> <p> /**<br /> * 如果发生与EditLog日志相关的IO异常，会从editStreams列表中删除掉发生IO异常的输出流对象<br /> * 第三个：对列表editStreams中的输出流对象进行批量检查并对发生IO异常的输出流对象处理<br /> */<br /> private void processIOError(ArrayList<EditLogOutputStream> errorStreams);

在FSEditLog类中处理IO异常的时候，还需要将处理的情况报告到FSImage中，因为FSImage也维护了一个指定为要删除的存储目录列表，保证数据状态的同步与一致。

4、加载EditLog日志文件

实现的方法为loadFSEdits方法，因为加载一个EditLog日志文件的时候，需要将对应EditLog日志文件记录内容应用到一个滞留于内存中的结构上，保证内存中对应结构与日志同步。因为该方法比较重要，通过它能够看到如何在日志文件与其对应的内存映像之间进行切换，尽管该方法代码比较多。我还是贴出来分析，能够更好地看到这个过程，加深理解。必要的时候我会对代码行修改，减少显示行数。

loadFSEdits方法实现如下所示：

static int loadFSEdits(EditLogInputStream edits) throws IOException {<br /> FSNamesystem fsNamesys = FSNamesystem.getFSNamesystem(); // 获取到一个FSNamesystem对象<br /> FSDirectory fsDir = fsNamesys.dir; // 得到一个fsNamesys.dir的引用<br /> int numEdits = 0, logVersion = 0;<br /> String clientName = null, clientMachine = null, path = null;<br /> // 这是一组统计变量<br /> int numOpAdd = 0, numOpClose = 0, numOpDelete = 0, numOpRename = 0, numOpSetRepl = 0, numOpMkDir = 0,<br /> numOpSetPerm = 0, numOpSetOwner = 0, numOpSetGenStamp = 0, numOpTimes = 0, numOpOther = 0;<br /> long startTime = FSNamesystem.now(); // EditLog日志文件开始加载的时间</p> <p> DataInputStream in = new DataInputStream(new BufferedInputStream(edits)); // EditLog日志文件对应的输入流<br /> try {<br /> in.mark(4); // 设置EditLog日志文件输入流当前位置，从而读取日志文件的版本（很可能发生版本号信息丢失的情况）<br /> // 如果EditLog日志文件大于2G，调用available方法将会返回一个负数，为了避免不得不调用available方法的情况，设置一个Boolean变量，并指定其值为true<br /> boolean available = true;<br /> try {<br /> logVersion = in.readByte(); // 从输入流中读取版本号信息<br /> } catch (EOFException e) {<br /> available = false; // 发生异常则置于available为false，表示当前EditLog日志文件不可用（可能日志文件存在问题）<br /> }<br /> if (available) { // 日志文件可用<br /> in.reset(); // 重置<br /> logVersion = in.readInt(); // 读取版本号<br /> if (logVersion < FSConstants.LAYOUT_VERSION) // 判断读取到的版本号是否合法<br /> throw new IOException("Unexpected version of the file system log file: " + logVersion + ". Current version = " + FSConstants.LAYOUT_VERSION + ".");<br /> }<br /> assert logVersion <= Storage.LAST_UPGRADABLE_LAYOUT_VERSION : "Unsupported version " + logVersion;</p> <p> while (true) {<br /> long timestamp = 0, mtime = 0, atime = 0, blockSize = 0;<br /> byte opcode = -1;<br /> try {<br /> opcode = in.readByte(); // 读取一个字节<br /> if (opcode == OP_INVALID) { // 因为我们已经知道，在开始写日志的时候，写完版本号以后，可能写入一个字节的OP_INVALID（也就是-1）；并且在写日志过程中，会将这个标记删除掉的，如果没有及时删除掉，很可能在该日志上根本没有事务记录，也可能发生了IO异常<br /> FSNamesystem.LOG.info("Invalid opcode, reached end of edit log " + "Number of transactions found " + numEdits);<br /> break; // 没有事务记录，直接退出，不再读取EditLog日志文件了<br /> }<br /> } catch (EOFException e) {<br /> break; // 没有事务记录<br /> }<br /> numEdits++; // 每读取到EditLog日志文件中一个记录的事务，就增加一个计数<br /> switch (opcode) { // 根据读取到的操作代码，进行相应的处理<br /> case OP_ADD:<br /> case OP_CLOSE: { // OP_ADD（添加）或OP_CLOSE（关闭）文件操作<br /> int length = in.readInt();<br /> if (-7 == logVersion && length != 3|| -17 < logVersion && logVersion < -7 && length != 4 || logVersion <= -17 && length != 5) { // 判断读取到的版本号是否合法<br /> throw new IOException("Incorrect data format." + " logVersion is " + logVersion + " but writables.length is " + length + ". ");<br /> }<br /> path = FSImage.readString(in); // 读取文件路径<br /> short replication = adjustReplication(readShort(in)); // 读取副本数（必要的话需要修改该值）<br /> mtime = readLong(in); // 读取文件修改时间<br /> if (logVersion <= -17) {<br /> atime = readLong(in); // 如果版本号小于-17，读取文件访问时间<br /> }<br /> if (logVersion < -7) {<br /> blockSize = readLong(in); // 读取块大小<br /> }<br /> Block blocks[] = null;<br /> if (logVersion <= -14) {<br /> blocks = readBlocks(in); // 读取块列表<br /> } else {<br /> BlockTwo oldblk = new BlockTwo();<br /> int num = in.readInt();<br /> blocks = new Block[num];<br /> for (int i = 0; i < num; i++) {<br /> oldblk.readFields(in);<br /> blocks[i] = new Block(oldblk.blkid, oldblk.len, Block.GRANDFATHER_GENERATION_STAMP);<br /> }<br /> }</p> <p> // 旧版本HDFS没有在文件中存储块大小，如果某个文件块多于1个，使用第一个作为块大小，否则使用默认值<br /> if (-8 <= logVersion && blockSize == 0) {<br /> if (blocks.length > 1) {<br /> blockSize = blocks[0].getNumBytes();<br /> } else {<br /> long first = ((blocks.length == 1)? blocks[0].getNumBytes(): 0);<br /> blockSize = Math.max(fsNamesys.getDefaultBlockSize(), first);<br /> }<br /> }</p> <p> PermissionStatus permissions = fsNamesys.getUpgradePermission();<br /> if (logVersion <= -11) {<br /> permissions = PermissionStatus.read(in); // 读取权限<br /> }</p> <p> // 读取文件中最后一个块的clientname, clientMachine and block locations<br /> if (opcode == OP_ADD && logVersion <= -12) {<br /> clientName = FSImage.readString(in);<br /> clientMachine = FSImage.readString(in);<br /> if (-13 <= logVersion) {<br /> readDatanodeDescriptorArray(in);<br /> }<br /> } else {<br /> clientName = "";<br /> clientMachine = "";<br /> }</p> <p> // The open lease transaction re-creates a file if necessary.<br /> if (FSNamesystem.LOG.isDebugEnabled()) {<br /> FSNamesystem.LOG.debug(opcode + ": " + path + " numblocks : " + blocks.length + " clientHolder " + clientName + " clientMachine " + clientMachine);<br /> }<br /> fsDir.unprotectedDelete(path, mtime); // 从namespace中删除该文件，并更新目录配额信息</p> <p> // 将path添加到文件树fsDir中<br /> INodeFile node = (INodeFile)fsDir.unprotectedAddFile(path, permissions, blocks, replication, mtime, atime, blockSize);<br /> if (opcode == OP_ADD) { // 如果是添加操作码<br /> numOpAdd++; // 统计<br /> INodeFileUnderConstruction cons = new INodeFileUnderConstruction(<br /> node.getLocalNameBytes(), node.getReplication(), node.getModificationTime(),<br /> node.getPreferredBlockSize(), node.getBlocks(), node.getPermissionStatus(),<br /> clientName, clientMachine, null);<br /> fsDir.replaceNode(path, node, cons); // 使用cons替换node<br /> fsNamesys.leaseManager.addLease(cons.clientName, path); // 将cons加载到内存中，加入到租约管理器中<br /> }<br /> break;<br /> }<br /> case OP_SET_REPLICATION: { // 设置副本数操作<br /> numOpSetRepl++; // 统计<br /> path = FSImage.readString(in); // 读取文件<br /> short replication = adjustReplication(readShort(in)); // 读取并调整副本数<br /> fsDir.unprotectedSetReplication(path, replication, null); // 更新目录fsDir<br /> break;<br /> }<br /> case OP_RENAME: { // 重命名操作<br /> numOpRename++; // 统计<br /> int length = in.readInt(); // 读取长度<br /> if (length != 3) {<br /> throw new IOException("Incorrect data format. " + "Mkdir operation.");<br /> }<br /> String s = FSImage.readString(in); // 读取源文件名称<br /> String d = FSImage.readString(in); // 读取重命名后文件名称<br /> timestamp = readLong(in); // 读取重命名文件时间戳<br /> FileStatus dinfo = fsDir.getFileInfo(d); // 获取文件d的描述信息<br /> fsDir.unprotectedRenameTo(s, d, timestamp); // 更新目录fsDir<br /> fsNamesys.changeLease(s, d, dinfo); // 修改租约信息<br /> break;<br /> }<br /> case OP_DELETE: { // 删除操作<br /> numOpDelete++; // 统计<br /> int length = in.readInt(); // 删除文件名称长度<br /> if (length != 2) {<br /> throw new IOException("Incorrect data format. " + "delete operation.");<br /> }<br /> path = FSImage.readString(in); // 读取长度<br /> timestamp = readLong(in); // 读取删除文件时间戳<br /> fsDir.unprotectedDelete(path, timestamp); // 执行删除<br /> break;<br /> }<br /> case OP_MKDIR: { // 创建目录操作<br /> numOpMkDir++; // 统计<br /> PermissionStatus permissions = fsNamesys.getUpgradePermission(); // 获取文件系统中路径的默认权限<br /> int length = in.readInt();<br /> if (-17 < logVersion && length != 2 || <= -17 && length != 3) {<br /> throw new IOException("Incorrect data format. " + "Mkdir operation.");<br /> }<br /> path = FSImage.readString(in); // 读取目录名称<br /> timestamp = readLong(in); // 读取创建目录时间戳</p> <p> if (logVersion <= -17) {<br /> atime = readLong(in);<br /> }</p> <p> if (logVersion <= -11) {<br /> permissions = PermissionStatus.read(in);<br /> }<br /> fsDir.unprotectedMkdir(path, permissions, timestamp); // 在fsDir目录中创建目录path<br /> break;<br /> }<br /> case OP_SET_GENSTAMP: { // 设置时间戳操作<br /> numOpSetGenStamp++; // 统计<br /> long lw = in.readLong(); // 读取设置的时间戳<br /> fsDir.namesystem.setGenerationStamp(lw); // 在fsDir中设置时间戳<br /> break;<br /> }<br /> case OP_DATANODE_ADD: { // 丢弃的操作码<br /> numOpOther++;<br /> FSImage.DatanodeImage nodeimage = new FSImage.DatanodeImage();<br /> nodeimage.readFields(in);<br /> break;<br /> }<br /> case OP_DATANODE_REMOVE: { // 丢弃的操作码<br /> numOpOther++;<br /> DatanodeID nodeID = new DatanodeID();<br /> nodeID.readFields(in);<br /> break;<br /> }<br /> case OP_SET_PERMISSIONS: { // 设置权限<br /> numOpSetPerm++; // 统计<br /> if (logVersion > -11)<br /> throw new IOException("Unexpected opcode " + opcode + " for version " + logVersion);<br /> fsDir.unprotectedSetPermission(FSImage.readString(in), FsPermission.read(in)); // 在fsDir中同步设置<br /> break;<br /> }<br /> case OP_SET_OWNER: { // 设置属主操作<br /> numOpSetOwner++;<br /> if (logVersion > -11)<br /> throw new IOException("Unexpected opcode " + opcode + " for version " + logVersion);<br /> fsDir.unprotectedSetOwner(FSImage.readString(in), FSImage.readString_EmptyAsNull(in), FSImage.readString_EmptyAsNull(in)); // 在fsDir中同步设置属主<br /> break;<br /> }<br /> case OP_SET_NS_QUOTA: { // 设置namespace配额操作<br /> if (logVersion > -16) {<br /> throw new IOException("Unexpected opcode " + opcode + " for version " + logVersion);<br /> }<br /> fsDir.unprotectedSetQuota(FSImage.readString(in), readLongWritable(in), FSConstants.QUOTA_DONT_SET); // 在fsDir中同步设置namespace配额<br /> break;<br /> }<br /> case OP_CLEAR_NS_QUOTA: { // 清除namespace配额操作<br /> if (logVersion > -16) {<br /> throw new IOException("Unexpected opcode " + opcode + " for version " + logVersion);<br /> }<br /> fsDir.unprotectedSetQuota(FSImage.readString(in), FSConstants.QUOTA_RESET, FSConstants.QUOTA_DONT_SET); // 在fsDir中同步清除namespace配额<br /> break;<br /> }<br /> case OP_SET_QUOTA: // 设置名称和磁盘配额<br /> fsDir.unprotectedSetQuota(FSImage.readString(in), readLongWritable(in), readLongWritable(in)); // 在fsDir中同步设置磁盘配额<br /> break;<br /> case OP_TIMES: { // 设置文件的mod & access时间操作<br /> numOpTimes++;<br /> int length = in.readInt();<br /> if (length != 3) {<br /> throw new IOException("Incorrect data format. " + "times operation.");<br /> }<br /> path = FSImage.readString(in);<br /> mtime = readLong(in);<br /> atime = readLong(in);<br /> fsDir.unprotectedSetTimes(path, mtime, atime, true); // 在fsDir上同步设置<br /> break;<br /> }<br /> default: {<br /> throw new IOException("Never seen opcode " + opcode);<br /> }<br /> }<br /> }<br /> } finally {<br /> in.close();<br /> }<br /> FSImage.LOG.info("Edits file " + edits.getName() + " of size " + edits.length() + " edits # " + numEdits + " loaded in " + (FSNamesystem.now()-startTime)/1000 + " seconds.");</p> <p> if (FSImage.LOG.isDebugEnabled()) {<br /> FSImage.LOG.debug("numOpAdd = " + numOpAdd + " numOpClose = " + numOpClose<br /> + " numOpDelete = " + numOpDelete + " numOpRename = " + numOpRename<br /> + " numOpSetRepl = " + numOpSetRepl + " numOpMkDir = " + numOpMkDir<br /> + " numOpSetPerm = " + numOpSetPerm<br /> + " numOpSetOwner = " + numOpSetOwner<br /> + " numOpSetGenStamp = " + numOpSetGenStamp<br /> + " numOpTimes = " + numOpTimes<br /> + " numOpOther = " + numOpOther);<br /> }</p> <p> if (logVersion != FSConstants.LAYOUT_VERSION) // 如果版本号不等于LAYOUT_VERSION=-8<br /> numEdits++; // 也进行统计<br /> return numEdits;<br /> }

通过该方法的实现，实际上在从多个EditLog 中读取日志信息的时候，主要是将日志文件中对应的事务，通过FSDirectory fsDir = fsNamesys.dir及时更新到文件系统的目录上，保持数据状态同步一致。

5、其它方法

还有几个方法，分别表示对EditLog日志文件的操作，比如：

关闭日志文件方法rollEditLog，同时创建一个edits.new文件；

删除旧日志文件purgeEditLog方法，同时将edits.new文件重命名为edits；

同步日志文件方法logSync，只对当前线程对日志文件作出的全部修改，同步到日志文件上。