一、前言
前一篇已经分析了序列化，这篇接着分析Zookeeper的持久化过程源码，持久化对于数据的存储至关重要，下面进行详细分析。
二、持久化总体框架
持久化的类主要在包org.apache.zookeeper.server.persistence下，此次也主要是对其下的类进行分析，其包下总体的类结构如下图所示。
· TxnLog，接口类型，读取事务性日志的接口。
· FileTxnLog，实现TxnLog接口，添加了访问该事务性日志的API。
· Snapshot，接口类型，持久层快照接口。
· FileSnap，实现Snapshot接口，负责存储、序列化、反序列化、访问快照。
· FileTxnSnapLog，封装了TxnLog和SnapShot。
· Util，工具类，提供持久化所需的API。
下面先来分析TxnLog和FileTxnLog的源码。
三、TxnLog源码分析
TxnLog是接口，规定了对日志的响应操作。
public interface TxnLog { /** * roll the current * log being appended to * @throws IOException */ // 回滚日志 void rollLog() throws IOException; /** * Append a request to the transaction log * @param hdr the transaction header * @param r the transaction itself * returns true iff something appended, otw false * @throws IOException */ // 添加一个请求至事务性日志 boolean append(TxnHeader hdr, Record r) throws IOException; /** * Start reading the transaction logs * from a given zxid * @param zxid * @return returns an iterator to read the * next transaction in the logs. * @throws IOException */ // 读取事务性日志 TxnIterator read(long zxid) throws IOException; /** * the last zxid of the logged transactions. * @return the last zxid of the logged transactions. * @throws IOException */ // 事务性操作的最新zxid long getLastLoggedZxid() throws IOException; /** * truncate the log to get in sync with the * leader. * @param zxid the zxid to truncate at. * @throws IOException */ // 清空日志，与Leader保持同步 boolean truncate(long zxid) throws IOException; /** * the dbid for this transaction log. * @return the dbid for this transaction log. * @throws IOException */ // 获取数据库的id long getDbId() throws IOException; /** * commmit the trasaction and make sure * they are persisted * @throws IOException */ // 提交事务并进行确认 void commit() throws IOException; /** * close the transactions logs */ // 关闭事务性日志 void close() throws IOException; /** * an iterating interface for reading * transaction logs. */ // 读取事务日志的迭代器接口 public interface TxnIterator { /** * return the transaction header. * @return return the transaction header. */ // 获取事务头部 TxnHeader getHeader(); /** * return the transaction record. * @return return the transaction record. */ // 获取事务 Record getTxn(); /** * go to the next transaction record. * @throws IOException */ // 下个事务 boolean next() throws IOException; /** * close files and release the * resources * @throws IOException */ // 关闭文件释放资源 void close() throws IOException; } }
其中，TxnLog除了提供读写事务日志的API外，还提供了一个用于读取日志的迭代器接口TxnIterator。
四、FileTxnLog源码分析
对于LogFile而言，其格式可分为如下三部分
LogFile:
FileHeader TxnList ZeroPad
FileHeader格式如下
FileHeader: {
magic 4bytes (ZKLG)
version 4bytes
dbid 8bytes
}
TxnList格式如下
TxnList:
Txn || Txn TxnList
Txn格式如下
Txn:
checksum Txnlen TxnHeader Record 0x42
Txnlen格式如下
Txnlen:
len 4bytes
TxnHeader格式如下
TxnHeader: {
sessionid 8bytes
cxid 4bytes
zxid 8bytes
time 8bytes
type 4bytes
}
ZeroPad格式如下
ZeroPad:
0 padded to EOF (filled during preallocation stage)
了解LogFile的格式对于理解源码会有很大的帮助。
4.1 属性
public class FileTxnLog implements TxnLog { private static final Logger LOG; // 预分配大小 64M static long preAllocSize = 65536 * 1024; // 魔术数字，默认为1514884167 public final static int TXNLOG_MAGIC = ByteBuffer.wrap("ZKLG".getBytes()).getInt(); // 版本号 public final static int VERSION = 2; /** Maximum time we allow for elapsed fsync before WARNing */ // 进行同步时，发出warn之前所能等待的最长时间 private final static long fsyncWarningThresholdMS; // 静态属性，确定Logger、预分配空间大小和最长时间 static { LOG = LoggerFactory.getLogger(FileTxnLog.class); String size = System.getProperty("zookeeper.preAllocSize"); if (size != null) { try { preAllocSize = Long.parseLong(size) * 1024; } catch (NumberFormatException e) { LOG.warn(size + " is not a valid value for preAllocSize"); } } fsyncWarningThresholdMS = Long.getLong("fsync.warningthresholdms", 1000); } // 最大(新)的zxid long lastZxidSeen; // 存储数据相关的流 volatile BufferedOutputStream logStream = null; volatile OutputArchive oa; volatile FileOutputStream fos = null; // log目录文件 File logDir; // 是否强制同步 private final boolean forceSync = !System.getProperty("zookeeper.forceSync", "yes").equals("no");; // 数据库id long dbId; // 流列表 private LinkedList<FileOutputStream> streamsToFlush = new LinkedList<FileOutputStream>(); // 当前大小 long currentSize; // 写日志文件 File logFileWrite = null; }
4.2. 核心函数
1. append函数
public synchronized boolean append(TxnHeader hdr, Record txn) throws IOException { if (hdr != null) { // 事务头部不为空 if (hdr.getZxid() <= lastZxidSeen) { // 事务的zxid小于等于最后的zxid LOG.warn("Current zxid " + hdr.getZxid() + " is <= " + lastZxidSeen + " for " + hdr.getType()); } if (logStream==null) { // 日志流为空 if(LOG.isInfoEnabled()){ LOG.info("Creating new log file: log." + Long.toHexString(hdr.getZxid())); } // logFileWrite = new File(logDir, ("log." + Long.toHexString(hdr.getZxid()))); fos = new FileOutputStream(logFileWrite); logStream=new BufferedOutputStream(fos); oa = BinaryOutputArchive.getArchive(logStream); // FileHeader fhdr = new FileHeader(TXNLOG_MAGIC,VERSION, dbId); // 序列化 fhdr.serialize(oa, "fileheader"); // Make sure that the magic number is written before padding. // 刷新到磁盘 logStream.flush(); // 当前通道的大小 currentSize = fos.getChannel().position(); // 添加fos streamsToFlush.add(fos); } // 填充文件 padFile(fos); // Serializes transaction header and transaction data into a byte buffer. // 将事务头和事务数据序列化成Byte Buffer byte[] buf = Util.marshallTxnEntry(hdr, txn); if (buf == null || buf.length == 0) { // 为空，抛出异常 throw new IOException("Faulty serialization for header " + "and txn"); } // 生成一个验证算法 Checksum crc = makeChecksumAlgorithm(); // Updates the current checksum with the specified array of bytes // 使用Byte数组来更新当前的Checksum crc.update(buf, 0, buf.length); // 写long类型数据 oa.writeLong(crc.getValue(), "txnEntryCRC"); // Write the serialized transaction record to the output archive. // 将序列化的事务记录写入OutputArchive Util.writeTxnBytes(oa, buf); return true; } return false; }
说明：append函数主要用做向事务日志中添加一个条目，其大体步骤如下
① 检查TxnHeader是否为空，若不为空，则进入②，否则，直接返回false
② 检查logStream是否为空(初始化为空)，若不为空，则进入③，否则，进入⑤
③ 初始化写数据相关的流和FileHeader，并序列化FileHeader至指定文件，进入④
④ 强制刷新（保证数据存到磁盘），并获取当前写入数据的大小。进入⑤
⑤ 填充数据，填充0，进入⑥
⑥ 将事务头和事务序列化成ByteBuffer（使用Util.marshallTxnEntry函数），进入⑦
⑦ 使用Checksum算法更新步骤⑥的ByteBuffer。进入⑧
⑧ 将更新的ByteBuffer写入磁盘文件，返回true
append间接调用了padLog函数，其源码如下
public static long padLogFile(FileOutputStream f,long currentSize, long preAllocSize) throws IOException{ // 获取位置 long position = f.getChannel().position(); if (position + 4096 >= currentSize) { // 计算后是否大于当前大小 // 重新设置当前大小，剩余部分填充0 currentSize = currentSize + preAllocSize; fill.position(0); f.getChannel().write(fill, currentSize-fill.remaining()); } return currentSize; }
说明：其主要作用是当文件大小不满64MB时，向文件填充0以达到64MB大小。
2. getLogFiles函数
public static File[] getLogFiles(File[] logDirList,long snapshotZxid) { // 按照zxid对文件进行排序 List<File> files = Util.sortDataDir(logDirList, "log", true); long logZxid = 0; // Find the log file that starts before or at the same time as the // zxid of the snapshot for (File f : files) { // 遍历文件 // 从文件中获取zxid long fzxid = Util.getZxidFromName(f.getName(), "log"); if (fzxid > snapshotZxid) { // 跳过大于snapshotZxid的文件 continue; } // the files // are sorted with zxid's if (fzxid > logZxid) { // 找出文件中最大的zxid(同时还需要小于等于snapshotZxid) logZxid = fzxid; } } // 文件列表 List<File> v=new ArrayList<File>(5); for (File f : files) { // 再次遍历文件 // 从文件中获取zxid long fzxid = Util.getZxidFromName(f.getName(), "log"); if (fzxid < logZxid) { // 跳过小于logZxid的文件 continue; } // 添加 v.add(f); } // 转化成File[] 类型后返回 return v.toArray(new File[0]); }
说明：该函数的作用是找出刚刚小于或者等于snapshot的所有log文件。其步骤大致如下。
① 对所有log文件按照zxid进行升序排序，进入②
② 遍历所有log文件并记录刚刚小于或等于给定snapshotZxid的log文件的logZxid，进入③
③ 再次遍历log文件，添加zxid大于等于步骤②中的logZxid的所有log文件，进入④
④ 转化后返回
getLogFiles函数调用了sortDataDir，其源码如下：
public static List<File> sortDataDir(File[] files, String prefix, boolean ascending) { if(files==null) return new ArrayList<File>(0); // 转化为列表 List<File> filelist = Arrays.asList(files); // 进行排序，Comparator是关键，根据zxid进行排序 Collections.sort(filelist, new DataDirFileComparator(prefix, ascending)); return filelist; }
说明：其用于排序log文件，可以选择根据zxid进行升序或降序。
getLogFiles函数间接调用了getZxidFromName，其源码如下：
// 从文件名中解析出zxid public static long getZxidFromName(String name, String prefix) { long zxid = -1; // 对文件名进行分割 String nameParts[] = name.split("\\."); if (nameParts.length == 2 && nameParts[0].equals(prefix)) { // 前缀相同 try { // 转化成长整形 zxid = Long.parseLong(nameParts[1], 16); } catch (NumberFormatException e) { } } return zxid; }
说明：getZxidFromName主要用作从文件名中解析zxid，并且需要从指定的前缀开始。
3. getLastLoggedZxid函数
public long getLastLoggedZxid() { // 获取已排好序的所有的log文件 File[] files = getLogFiles(logDir.listFiles(), 0); // 获取最大的zxid(最后一个log文件对应的zxid) long maxLog=files.length>0? Util.getZxidFromName(files[files.length-1].getName(),"log"):-1; // if a log file is more recent we must scan it to find // the highest zxid // long zxid = maxLog; // 迭代器 TxnIterator itr = null; try { // 新生FileTxnLog FileTxnLog txn = new FileTxnLog(logDir); // 开始读取从给定zxid之后的所有事务 itr = txn.read(maxLog); while (true) { // 遍历 if(!itr.next()) // 是否存在下一项 break; // 获取事务头 TxnHeader hdr = itr.getHeader(); // 获取zxid zxid = hdr.getZxid(); } } catch (IOException e) { LOG.warn("Unexpected exception", e); } finally { // 关闭迭代器 close(itr); } return zxid; }
说明：该函数主要用于获取记录在log中的最后一个zxid。其步骤大致如下
① 获取已排好序的所有log文件，并从最后一个文件中取出zxid作为候选的最大zxid，进入②
② 新生成FileTxnLog并读取步骤①中zxid之后的所有事务，进入③
③ 遍历所有事务并提取出相应的zxid，最后返回。
其中getLastLoggedZxid调用了read函数，其源码如下
public TxnIterator read(long zxid) throws IOException { // 返回事务文件访问迭代器 return new FileTxnIterator(logDir, zxid); }
说明：read函数会生成一个FileTxnIterator，其是TxnLog.TxnIterator的子类，之后在FileTxnIterator构造函数中会调用init函数，其源码如下
void init() throws IOException { // 新生成文件列表 storedFiles = new ArrayList<File>(); // 进行排序 List<File> files = Util.sortDataDir(FileTxnLog.getLogFiles(logDir.listFiles(), 0), "log", false); for (File f: files) { // 遍历文件 if (Util.getZxidFromName(f.getName(), "log") >= zxid) { // 添加zxid大于等于指定zxid的文件 storedFiles.add(f); } // add the last logfile that is less than the zxid else if (Util.getZxidFromName(f.getName(), "log") < zxid) { // 只添加一个zxid小于指定zxid的文件，然后退出 storedFiles.add(f); break; } } // go to the next logfile // 进入下一个log文件 goToNextLog(); if (!next()) // 不存在下一项，返回 return; while (hdr.getZxid() < zxid) { // 从事务头中获取zxid小于给定zxid，直到不存在下一项或者大于给定zxid时退出 if (!next()) return; } }
说明：init函数用于进行初始化操作，会根据zxid的不同进行不同的初始化操作，在init函数中会调用goToNextLog函数，其源码如下
private boolean goToNextLog() throws IOException { if (storedFiles.size() > 0) { // 存储的文件列表大于0 // 取最后一个log文件 this.logFile = storedFiles.remove(storedFiles.size()-1); // 针对该文件，创建InputArchive ia = createInputArchive(this.logFile); // 返回true return true; } return false; }
说明：goToNextLog表示选取下一个log文件，在init函数中还调用了next函数，其源码如下
public boolean next() throws IOException { if (ia == null) { // 为空，返回false return false; } try { // 读取长整形crcValue long crcValue = ia.readLong("crcvalue"); // 通过input archive读取一个事务条目 byte[] bytes = Util.readTxnBytes(ia); // Since we preallocate, we define EOF to be an if (bytes == null || bytes.length==0) { // 对bytes进行判断 throw new EOFException("Failed to read " + logFile); } // EOF or corrupted record // validate CRC // 验证CRC Checksum crc = makeChecksumAlgorithm(); // 更新 crc.update(bytes, 0, bytes.length); if (crcValue != crc.getValue()) // 验证不相等，抛出异常 throw new IOException(CRC_ERROR); if (bytes == null || bytes.length == 0) // bytes为空，返回false return false; // 新生成TxnHeader hdr = new TxnHeader(); // 将Txn反序列化，并且将对应的TxnHeader反序列化至hdr，整个Record反序列化至record record = SerializeUtils.deserializeTxn(bytes, hdr); } catch (EOFException e) { // 抛出异常 LOG.debug("EOF excepton " + e); // 关闭输入流 inputStream.close(); // 赋值为null inputStream = null; ia = null; hdr = null; // this means that the file has ended // we should go to the next file if (!goToNextLog()) { // 没有log文件，则返回false return false; } // if we went to the next log file, we should call next() again // 继续调用next return next(); } catch (IOException e) { inputStream.close(); throw e; } // 返回true return true; }
说明：next表示将迭代器移动至下一个事务，方便读取，next函数的步骤如下。
① 读取事务的crcValue值，用于后续的验证，进入②
② 读取事务，使用CRC32进行更新并与①中的结果进行比对，若不相同，则抛出异常，否则，进入③
③ 将事务进行反序列化并保存至相应的属性中（如事务头和事务体），会确定具体的事务操作类型。
④ 在读取过程抛出异常时，会首先关闭流，然后再尝试调用next函数（即进入下一个事务进行读取）。
4. commit函数
public synchronized void commit() throws IOException { if (logStream != null) { // 强制刷到磁盘 logStream.flush(); } for (FileOutputStream log : streamsToFlush) { // 遍历流 // 强制刷到磁盘 log.flush(); if (forceSync) { // 是否强制同步 long startSyncNS = System.nanoTime(); log.getChannel().force(false); // 计算流式的时间 long syncElapsedMS = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS); if (syncElapsedMS > fsyncWarningThresholdMS) { // 大于阈值时则会警告 LOG.warn("fsync-ing the write ahead log in " + Thread.currentThread().getName() + " took " + syncElapsedMS + "ms which will adversely effect operation latency. " + "See the ZooKeeper troubleshooting guide"); } } } while (streamsToFlush.size() > 1) { // 移除流并关闭 streamsToFlush.removeFirst().close(); } }
说明：该函数主要用于提交事务日志至磁盘，其大致步骤如下
① 若日志流logStream不为空，则强制刷新至磁盘，进入②
② 遍历需要刷新至磁盘的所有流streamsToFlush并进行刷新，进入③
③ 判断是否需要强制性同步，如是，则计算每个流的流式时间并在控制台给出警告，进入④
④ 移除所有流并关闭。
5. truncate函数
public boolean truncate(long zxid) throws IOException { FileTxnIterator itr = null; try { // 获取迭代器 itr = new FileTxnIterator(this.logDir, zxid); PositionInputStream input = itr.inputStream; long pos = input.getPosition(); // now, truncate at the current position // 从当前位置开始清空 RandomAccessFile raf = new RandomAccessFile(itr.logFile, "rw"); raf.setLength(pos); raf.close(); while (itr.goToNextLog()) { // 存在下一个log文件 if (!itr.logFile.delete()) { // 删除 LOG.warn("Unable to truncate {}", itr.logFile); } } } finally { // 关闭迭代器 close(itr); } return true; }
说明：该函数用于清空大于给定zxid的所有事务日志。
五、总结
对于持久化中的TxnLog和FileTxnLog的源码分析就已经完成了，其源码还是相对简单，也谢谢各位园友的观看~
