zookeeper在将写请求应用到内存数据库之前，会首先记录这次事务，从而生成事务日志。当事务日志的数量达到一定规模之后，会对内存数据库打快照，从而生成快照文件。
持久化相关的类
TxnLog包含了对事务日志的操作方法
FileTnxLog实现了TxnLog接口
SnapShot包含了对快照日志的操作方法
FileSnapShot实现了SnapShot接口
FileTnxSnapLog包含TnxLog和SnapShot
事务日志
TxnLog
public interface TxnLog {
// 设置服务端的统计信息
void setServerStats(ServerStats serverStats);
// 滚动当前正在写的日志文件
void rollLog() throws IOException;
// 追加指定日志
boolean append(TxnHeader hdr, Record r) throws IOException;
// 返回可以从指定事务id开始读取之后事务日志的迭代器
TxnIterator read(long zxid) throws IOException;
// 获取最近记录的事务日志的事务id
long getLastLoggedZxid() throws IOException;
// 截取事务日志到指定的事务id
boolean truncate(long zxid) throws IOException;
// 返回当前事务日志所属的dbid
long getDbId() throws IOException;
// 提交日志，保证日志能够持久化
void commit() throws IOException;
/**
*
* @return transaction log's elapsed sync time in milliseconds
*/
long getTxnLogSyncElapsedTime();
// 关闭日志
void close() throws IOException;
// 读取事务日志的迭代器
public interface TxnIterator {
// 获取事务记录的头
TxnHeader getHeader();
// 获取事务记录的内容
Record getTxn();
// 移动到下一个日志
boolean next() throws IOException;
// 关闭文件
void close() throws IOException;
// 获取当前迭代器能够返回的日志预估大小
long getStorageSize() throws IOException;
}
}
文件组成
一个事务日志文件一共有三个部分：
FileHeader TxnList ZeroPad
FileHeader
FileHeader {
magic 4bytes
version 4bytes
dbid 8bytes
}
TxnList
TxnList存放事务记录，包含多个Txn
Txn {
checksum 8bytes
Txnlen 4bytes
TxnHeader {
sessionid 8bytes
cxid 4bytes
zxid 8bytes
time 8bytes
type 4bytes
}
Record
0x42
}
ZeroPad
在文件末尾添加0
FileTxnLog
主要属性
// 文件头部的magic数字
public final static int TXNLOG_MAGIC =
ByteBuffer.wrap("ZKLG".getBytes()).getInt();
// 上一个日志记录的事务id
long lastZxidSeen;
// 日志文件流
volatile BufferedOutputStream logStream = null;
volatile OutputArchive oa;
volatile FileOutputStream fos = null;
// 事务日志文件存放目录
File logDir;
// 当提交时是否确保同步到磁盘
private final boolean forceSync = !System.getProperty("zookeeper.forceSync", "yes").equals("no");
// 内存数据库id
long dbId;
// 等待flush的流
private LinkedList<FileOutputStream> streamsToFlush =
new LinkedList<FileOutputStream>();
内部类
PositionInputStream#
PositionInputStream是一个能够记录当前读入位置的输入流，实现方式就是每次从输入流中读取数据后，就会增加position
static class PositionInputStream extends FilterInputStream {
long position;
protected PositionInputStream(InputStream in) {
super(in);
position = 0;
}
@Override
public int read() throws IOException {
int rc = super.read();
if (rc > -1) {
position++;
}
return rc;
}
public int read(byte[] b) throws IOException {
int rc = super.read(b);
if (rc > 0) {
position += rc;
}
return rc;
}
@Override
public int read(byte[] b, int off, int len) throws IOException {
int rc = super.read(b, off, len);
if (rc > 0) {
position += rc;
}
return rc;
}
@Override
public long skip(long n) throws IOException {
long rc = super.skip(n);
if (rc > 0) {
position += rc;
}
return rc;
}
public long getPosition() {
return position;
}
@Override
public boolean markSupported() {
return false;
}
@Override
public void mark(int readLimit) {
throw new UnsupportedOperationException("mark");
}
@Override
public void reset() {
throw new UnsupportedOperationException("reset");
}
}
FileTxnIterator#
FileTxnIterator的作用就是从事务日志中读取事务记录
首先，会解析日志目录下的文件，提取文件名中的zxid作为当前日志包含事务日志中的最小事务id
然后，将zxid大于等于指定事务id以及小于zxid的最大zxid事务日志文件保存下下来，并根据zxid递减排序
public static class FileTxnIterator implements TxnLog.TxnIterator {
// 日志文件目录
File logDir;
// 当前记录的事务id
long zxid;
// 当前记录的头
TxnHeader hdr;
// 当前记录的内容
Record record;
File logFile;
InputArchive ia;
static final String CRC_ERROR="CRC check failed";
PositionInputStream inputStream=null;
// 包含有大于指定事务id的事务文件
private ArrayList<File> storedFiles;
public FileTxnIterator(File logDir, long zxid, boolean fastForward)
throws IOException {
this.logDir = logDir;
this.zxid = zxid;
// 找到可能包含指定事务id的所有日志，并且排序
init();
// 如果设置了fastForward，会快速定位到该zxid位置，否则会定位到可能包含该zxid的日志的头部
if (fastForward && hdr != null) {
while (hdr.getZxid() < zxid) {
if (!next())
break;
}
}
}
/**
* create an iterator over a transaction database directory
* @param logDir the transaction database directory
* @param zxid the zxid to start reading from
* @throws IOException
*/
public FileTxnIterator(File logDir, long zxid) throws IOException {
this(logDir, zxid, true);
}
void init() throws IOException {
storedFiles = new ArrayList<File>();
// 根据文件名中的zxid进行递减排序
List<File> files = Util.sortDataDir(FileTxnLog.getLogFiles(logDir.listFiles(), 0), LOG_FILE_PREFIX, false);
// 筛选出大于指定zxid的日志和第一个小于zxid的日志文件
for (File f: files) {
if (Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX) >= zxid) {
storedFiles.add(f);
}
// add the last logfile that is less than the zxid
else if (Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX) < zxid) {
storedFiles.add(f);
break;
}
}
// 创建一个指向最后一个文件的输入流
goToNextLog();
// 定位到该文件中的第一条记录
next();
}
// 返回当前还没有处理的日志的预估大小
public long getStorageSize() {
long sum = 0;
for (File f : storedFiles) {
sum += f.length();
}
return sum;
}
// 移动到下一个日志文件，按照zxid递增的顺序来遍历文件
private boolean goToNextLog() throws IOException {
if (storedFiles.size() > 0) {
this.logFile = storedFiles.remove(storedFiles.size()-1);
ia = createInputArchive(this.logFile);
return true;
}
return false;
}
// 移动到下一个事务日志
public boolean next() throws IOException {
if (ia == null) {
return false;
}
try {
// 从日志文件输入流中恢复出日志记录
long crcValue = ia.readLong("crcvalue");
byte[] bytes = Util.readTxnBytes(ia);
// Since we preallocate, we define EOF to be an
if (bytes == null || bytes.length==0) {
throw new EOFException("Failed to read " + logFile);
}
// EOF or corrupted record
// validate CRC
Checksum crc = makeChecksumAlgorithm();
crc.update(bytes, 0, bytes.length);
if (crcValue != crc.getValue())
throw new IOException(CRC_ERROR);
hdr = new TxnHeader();
record = SerializeUtils.deserializeTxn(bytes, hdr);
} catch (EOFException e) {
// EOFException异常代表当前文件读取完毕，尝试调用toToNextLog切换到下一个文件
LOG.debug("EOF exception " + e);
inputStream.close();
inputStream = null;
ia = null;
hdr = null;
// this means that the file has ended
// we should go to the next file
if (!goToNextLog()) {
return false;
}
// if we went to the next log file, we should call next() again
return next();
} catch (IOException e) {
inputStream.close();
throw e;
}
return true;
}
/**
* return the current header
* @return the current header that
* is read
*/
public TxnHeader getHeader() {
return hdr;
}
/**
* return the current transaction
* @return the current transaction
* that is read
*/
public Record getTxn() {
return record;
}
/**
* close the iterator
* and release the resources.
*/
public void close() throws IOException {
if (inputStream != null) {
inputStream.close();
}
}
}
重要方法
append#
// FileTxnLog
public synchronized boolean append(TxnHeader hdr, Record txn)
throws IOException
{
if (hdr == null) {
return false;
}
// 如果当前的事务id小于日志中最大的事务id
// 从这里可以看出事务日志中的事务日志的事务id并不是一定递增的
if (hdr.getZxid() <= lastZxidSeen) {
LOG.warn("Current zxid " + hdr.getZxid()
+ " is <= " + lastZxidSeen + " for "
+ hdr.getType());
} else {
// 更新日志事务id的最大值
lastZxidSeen = hdr.getZxid();
}
// 当前没有创建日志文件
if (logStream==null) {
if(LOG.isInfoEnabled()){
LOG.info("Creating new log file: " + Util.makeLogName(hdr.getZxid()));
}
// 在logDir目录下创建log.当前事务记录id的日志文件
logFileWrite = new File(logDir, Util.makeLogName(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();
filePadding.setCurrentSize(fos.getChannel().position());
// 将当前流加入到需要flush的流列表中
streamsToFlush.add(fos);
}
filePadding.padFile(fos.getChannel());
// 将当前事务记录的头和内容序列化成字节数组
byte[] buf = Util.marshallTxnEntry(hdr, txn);
if (buf == null || buf.length == 0) {
throw new IOException("Faulty serialization for header " +
"and txn");
}
// 计算头部和内容的校验和
Checksum crc = makeChecksumAlgorithm();
crc.update(buf, 0, buf.length);
// 将校验和写入到文件中
oa.writeLong(crc.getValue(), "txnEntryCRC");
// 将事务头部、内容以及EOR写入到文件中
Util.writeTxnBytes(oa, buf);
return true;
}
上述方法首先判断当前是否有正在写的日志文件，如果没有会使用当前事务id生成log.事务id的事务日志文件
然后按照日志的格式，将当前事务记录写入到文件中
commit#
// FileTxnLog
public synchronized void commit() throws IOException {
// 当前有正在写入的日志文件，将数据刷写到操作系统，此时并不保证已经落盘
if (logStream != null) {
logStream.flush();
}
// 遍历streamToFlush，当前正在写入的流也已经加入到了streamsToFlush
for (FileOutputStream log : streamsToFlush) {
// 刷写到操作系统
log.flush();
// 如果需要保证落盘，开始落盘
if (forceSync) {
long startSyncNS = System.nanoTime();
FileChannel channel = log.getChannel();
channel.force(false);
// 记录当前文件落盘时间
syncElapsedMS = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS);
// 落盘时间超过指定的阈值，记录超出此时，并打印提示日志
if (syncElapsedMS > fsyncWarningThresholdMS) {
if(serverStats != null) {
serverStats.incrementFsyncThresholdExceedCount();
}
LOG.warn("fsync-ing the write ahead log in "
+ Thread.currentThread().getName()
+ " took " + syncElapsedMS
+ "ms which will adversely effect operation latency. "
+ "File size is " + channel.size() + " bytes. "
+ "See the ZooKeeper troubleshooting guide");
}
}
}
// 关闭streamsToFlush中的流，并清空streamsToFlush
while (streamsToFlush.size() > 1) {
streamsToFlush.removeFirst().close();
}
}
该方法的作用是保证生成的事务日志文件都刷写到操作系统中，如果设置了强制落盘，会落盘
read#
返回一个迭代器，该迭代器定位到了大于指定zxid的事务记录
public TxnIterator read(long zxid, boolean fastForward) throws IOException {
return new FileTxnIterator(logDir, zxid, fastForward);
}
rollLog#
// FileTxnLog
public synchronized void rollLog() throws IOException {
// flush当前写的事务日志文件流
if (logStream != null) {
this.logStream.flush();
this.logStream = null;
oa = null;
}
}
滚动当前日志
close#
// FileTxnLog
public synchronized void close() throws IOException {
// 关闭日志文件流
if (logStream != null) {
logStream.close();
}
for (FileOutputStream log : streamsToFlush) {
log.close();
}
}
truncate#
public boolean truncate(long zxid) throws IOException {
FileTxnIterator itr = null;
try {
itr = new FileTxnIterator(this.logDir, zxid);
PositionInputStream input = itr.inputStream;
if(input == null) {
throw new IOException("No log files found to truncate! This could " +
"happen if you still have snapshots from an old setup or " +
"log files were deleted accidentally or dataLogDir was changed in zoo.cfg.");
}
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()) {
if (!itr.logFile.delete()) {
LOG.warn("Unable to truncate {}", itr.logFile);
}
}
} finally {
close(itr);
}
return true;
}
快照日志
SnapShot
public interface SnapShot {
// 从最近的快照中恢复DateTree和session
long deserialize(DataTree dt, Map<Long, Integer> sessions)
throws IOException;
// 将当前的DataTree和session序列化到文件中
void serialize(DataTree dt, Map<Long, Integer> sessions,
File name)
throws IOException;
// 返回最近生成的快照文件
File findMostRecentSnapshot() throws IOException;
// 关闭序列化
void close() throws IOException;
}
FileSnap
FileSnap是SnapShot的实现
主要属性
// 快照文件目录
File snapDir;
// 是否关闭快照
private volatile boolean close = false;
// 版本
private static final int VERSION = 2;
private static final long dbId = -1;
private static final Logger LOG = LoggerFactory.getLogger(FileSnap.class);
// 快照文件头部的magic字段
public final static int SNAP_MAGIC
= ByteBuffer.wrap("ZKSN".getBytes()).getInt();
// 快照文件前缀
public static final String SNAPSHOT_FILE_PREFIX = "snapshot";
重要方法
serialize
// FileSnap
public synchronized void serialize(DataTree dt, Map<Long, Integer> sessions, File snapShot)
throws IOException {
if (!close) {
try (OutputStream sessOS = new BufferedOutputStream(new FileOutputStream(snapShot));
// CheckedOutputStream是带有校验的输出流，一边写一边更新校验值
CheckedOutputStream crcOut = new CheckedOutputStream(sessOS, new Adler32())) {
//CheckedOutputStream cout = new CheckedOutputStream()
OutputArchive oa = BinaryOutputArchive.getArchive(crcOut);
// 文件头
FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
// 实际进行序列化
serialize(dt, sessions, oa, header);
long val = crcOut.getChecksum().getValue();
// 写出校验值
oa.writeLong(val, "val");
oa.writeString("/", "path");
sessOS.flush();
}
}
}
// FileSnap
protected void serialize(DataTree dt,Map<Long, Integer> sessions,
OutputArchive oa, FileHeader header) throws IOException {
// this is really a programmatic error and not something that can
// happen at runtime
if(header==null)
throw new IllegalStateException(
"Snapshot's not open for writing: uninitialized header");
// 写入文件头
header.serialize(oa, "fileheader");
// 写入DataTree 和 session
SerializeUtils.serializeSnapshot(dt,oa,sessions);
}
// SerializeUtils
public static void serializeSnapshot(DataTree dt,OutputArchive oa,
Map<Long, Integer> sessions) throws IOException {
HashMap<Long, Integer> sessSnap = new HashMap<Long, Integer>(sessions);
// 写入当前事务的个数
oa.writeInt(sessSnap.size(), "count");
// 遍历事务，写入事务的id和过期时间
for (Entry<Long, Integer> entry : sessSnap.entrySet()) {
oa.writeLong(entry.getKey().longValue(), "id");
oa.writeInt(entry.getValue().intValue(), "timeout");
}
// 写入dataTree
dt.serialize(oa, "tree");
}
下面看下如何序列化DataTree
public void serialize(OutputArchive oa, String tag) throws IOException {
// 序列化aclCache
aclCache.serialize(oa);
// 序列化节点
serializeNode(oa, new StringBuilder(""));
// / marks end of stream
// we need to check if clear had been called in between the snapshot.
if (root != null) {
oa.writeString("/", "path");
}
}
void serializeNode(OutputArchive oa, StringBuilder path) throws IOException {
String pathString = path.toString();
// 获取路径对应的节点
DataNode node = getNode(pathString);
if (node == null) {
return;
}
String children[] = null;
// 拷贝当前节点
DataNode nodeCopy;
synchronized (node) {
StatPersisted statCopy = new StatPersisted();
copyStatPersisted(node.stat, statCopy);
//we do not need to make a copy of node.data because the contents
//are never changed
nodeCopy = new DataNode(node.data, node.acl, statCopy);
Set<String> childs = node.getChildren();
children = childs.toArray(new String[childs.size()]);
}
// 序列化节点
serializeNodeData(oa, pathString, nodeCopy);
path.append('/');
int off = path.length();
// 遍历子节点并进行序列化
for (String child : children) {
// since this is single buffer being resused
// we need
// to truncate the previous bytes of string.
path.delete(off, Integer.MAX_VALUE);
path.append(child);
serializeNode(oa, path);
}
}
下面看下如何序列化DataTree中的一个节点
public void serializeNodeData(OutputArchive oa, String path, DataNode node) throws IOException {
// 首先写入路径
oa.writeString(path, "path");
// 写入节点内容
oa.writeRecord(node, "node");
}
// DataNode
synchronized public void serialize(OutputArchive archive, String tag)
throws IOException {
archive.startRecord(this, "node");
// 写入数据
archive.writeBuffer(data, "data");
// 写入acl值
archive.writeLong(acl, "acl");
// 写入状态，比如修改时间，创建时间等
stat.serialize(archive, "statpersisted");
archive.endRecord(this, "node");
}
findNValidSnapshots
寻找最近的n个快照文件
private List<File> findNValidSnapshots(int n) throws IOException {
// 遍历快照日志下的文件，从文件名中提取事务id，按照事务id递减排序文件
List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
int count = 0;
List<File> list = new ArrayList<File>();
// 从这些文件中找到校验通过的n个文件
for (File f : files) {
// we should catch the exceptions
// from the valid snapshot and continue
// until we find a valid one
try {
if (Util.isValidSnapshot(f)) {
list.add(f);
count++;
if (count == n) {
break;
}
}
} catch (IOException e) {
LOG.info("invalid snapshot " + f, e);
}
}
return list;
}
findNRecentSnapshots
遍历快照目录下的快照文件，寻找最近的n个快照
public List<File> findNRecentSnapshots(int n) throws IOException {
List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
int count = 0;
List<File> list = new ArrayList<File>();
for (File f: files) {
if (count == n)
break;
if (Util.getZxidFromName(f.getName(), SNAPSHOT_FILE_PREFIX) != -1) {
count++;
list.add(f);
}
}
return list;
}
findMostRecentSnapshot
遍历快照目录下的快照文件，找到最近的快照文件
public File findMostRecentSnapshot() throws IOException {
List<File> files = findNValidSnapshots(1);
if (files.size() == 0) {
return null;
}
return files.get(0);
}
deserialize
public long deserialize(DataTree dt, Map<Long, Integer> sessions)
throws IOException {
// we run through 100 snapshots (not all of them)
// if we cannot get it running within 100 snapshots
// we should give up
// 找到最近有效的100个快照文件
// 这里找到的文件并不一定是完全有效的
List<File> snapList = findNValidSnapshots(100);
if (snapList.size() == 0) {
return -1L;
}
File snap = null;
boolean foundValid = false;
// 遍历找到的快照文件，读取文件，并且计算校验和，校验文件是否有效
// 序列化该文件到DateTree session
for (int i = 0, snapListSize = snapList.size(); i < snapListSize; i++) {
snap = snapList.get(i);
LOG.info("Reading snapshot " + snap);
try (InputStream snapIS = new BufferedInputStream(new FileInputStream(snap));
CheckedInputStream crcIn = new CheckedInputStream(snapIS, new Adler32())) {
InputArchive ia = BinaryInputArchive.getArchive(crcIn);
deserialize(dt, sessions, ia);
long checkSum = crcIn.getChecksum().getValue();
long val = ia.readLong("val");
if (val != checkSum) {
throw new IOException("CRC corruption in snapshot : " + snap);
}
foundValid = true;
break;
} catch (IOException e) {
LOG.warn("problem reading snap file " + snap, e);
}
}
if (!foundValid) {
throw new IOException("Not able to find valid snapshots in " + snapDir);
}
dt.lastProcessedZxid = Util.getZxidFromName(snap.getName(), SNAPSHOT_FILE_PREFIX);
return dt.lastProcessedZxid;
}
FileTxnSnapLog
主要属性
// 事务日志目录
private final File dataDir;
// 快照文件目录
private final File snapDir;
// 事务日志
private TxnLog txnLog;
// 快照
private SnapShot snapLog;
private final boolean trustEmptySnapshot;
public final static int VERSION = 2;
public final static String version = "version-";
private static final Logger LOG = LoggerFactory.getLogger(FileTxnSnapLog.class);
// 当目录不存在时是否自动创建目录
public static final String ZOOKEEPER_DATADIR_AUTOCREATE =
"zookeeper.datadir.autocreate";
public static final String ZOOKEEPER_DATADIR_AUTOCREATE_DEFAULT = "true";
重要方法
构造方法
// FileTxnSnapLog
public FileTxnSnapLog(File dataDir, File snapDir) throws IOException {
LOG.debug("Opening datadir:{} snapDir:{}", dataDir, snapDir);
// 事务日志文件目录和快照文件目录
// ${dataDir}/version-2
this.dataDir = new File(dataDir, version + VERSION);
this.snapDir = new File(snapDir, version + VERSION);
// 判断是否自动创建目录
boolean enableAutocreate = Boolean.valueOf(
System.getProperty(ZOOKEEPER_DATADIR_AUTOCREATE,
ZOOKEEPER_DATADIR_AUTOCREATE_DEFAULT));
trustEmptySnapshot = Boolean.getBoolean(ZOOKEEPER_SNAPSHOT_TRUST_EMPTY);
LOG.info(ZOOKEEPER_SNAPSHOT_TRUST_EMPTY + " : " + trustEmptySnapshot);
// 创建目录
if (!this.dataDir.exists()) {
if (!enableAutocreate) {
throw new DatadirException("Missing data directory "
+ this.dataDir
+ ", automatic data directory creation is disabled ("
+ ZOOKEEPER_DATADIR_AUTOCREATE
+ " is false). Please create this directory manually.");
}
if (!this.dataDir.mkdirs()) {
throw new DatadirException("Unable to create data directory "
+ this.dataDir);
}
}
if (!this.dataDir.canWrite()) {
throw new DatadirException("Cannot write to data directory " + this.dataDir);
}
if (!this.snapDir.exists()) {
// by default create this directory, but otherwise complain instead
// See ZOOKEEPER-1161 for more details
if (!enableAutocreate) {
throw new DatadirException("Missing snap directory "
+ this.snapDir
+ ", automatic data directory creation is disabled ("
+ ZOOKEEPER_DATADIR_AUTOCREATE
+ " is false). Please create this directory manually.");
}
if (!this.snapDir.mkdirs()) {
throw new DatadirException("Unable to create snap directory "
+ this.snapDir);
}
}
if (!this.snapDir.canWrite()) {
throw new DatadirException("Cannot write to snap directory " + this.snapDir);
}
// check content of transaction log and snapshot dirs if they are two different directories
// See ZOOKEEPER-2967 for more details
if(!this.dataDir.getPath().equals(this.snapDir.getPath())){
// 当事务日志目录和快照目录不同时，进行校验
// 判断日志目录下是否有快照文件
checkLogDir();
// 判断快照目录下是否有日志文件
checkSnapDir();
}
txnLog = new FileTxnLog(this.dataDir);
snapLog = new FileSnap(this.snapDir);
}
构造方法主要就是创建TxnLog和SnapLog
append
将请求透传给TxnLog
public boolean append(Request si) throws IOException {
return txnLog.append(si.getHdr(), si.getTxn());
}
save
使用DataTree中最新的事务id生成快照文件名，使用SnapLog来生成快照
public void save(DataTree dataTree,
ConcurrentHashMap<Long, Integer> sessionsWithTimeouts)
throws IOException {
long lastZxid = dataTree.lastProcessedZxid;
File snapshotFile = new File(snapDir, Util.makeSnapshotName(lastZxid));
LOG.info("Snapshotting: 0x{} to {}", Long.toHexString(lastZxid),
snapshotFile);
snapLog.serialize(dataTree, sessionsWithTimeouts, snapshotFile);
}
rollLog
public void rollLog() throws IOException {
txnLog.rollLog();
}
close
public void close() throws IOException {
txnLog.close();
snapLog.close();
}
truncateLog
关闭TxnLog和SnapShot，然后截断日志文件，重新创建TxnLog SnapShot
public boolean truncateLog(long zxid) throws IOException {
// close the existing txnLog and snapLog
close();
// truncate it
FileTxnLog truncLog = new FileTxnLog(dataDir);
boolean truncated = truncLog.truncate(zxid);
truncLog.close();
// re-open the txnLog and snapLog
// I'd rather just close/reopen this object itself, however that
// would have a big impact outside ZKDatabase as there are other
// objects holding a reference to this object.
txnLog = new FileTxnLog(dataDir);
snapLog = new FileSnap(snapDir);
return truncated;
}
restore
public long restore(DataTree dt, Map<Long, Integer> sessions,
PlayBackListener listener) throws IOException {
// 使用最近的有效快照文件来还原DataTree和session
long deserializeResult = snapLog.deserialize(dt, sessions);
FileTxnLog txnLog = new FileTxnLog(dataDir);
if (-1L == deserializeResult) {
/* this means that we couldn't find any snapshot, so we need to
* initialize an empty database (reported in ZOOKEEPER-2325) */
if (txnLog.getLastLoggedZxid() != -1) {
// ZOOKEEPER-3056: provides an escape hatch for users upgrading
// from old versions of zookeeper (3.4.x, pre 3.5.3).
if (!trustEmptySnapshot) {
throw new IOException(EMPTY_SNAPSHOT_WARNING + "Something is broken!");
} else {
LOG.warn(EMPTY_SNAPSHOT_WARNING + "This should only be allowed during upgrading.");
}
}
/* TODO: (br33d) we should either put a ConcurrentHashMap on restore()
* or use Map on save() */
save(dt, (ConcurrentHashMap<Long, Integer>)sessions);
/* return a zxid of zero, since we the database is empty */
return 0;
}
// 接着使用事务日志来恢复
return fastForwardFromEdits(dt, sessions, listener);
}
public long fastForwardFromEdits(DataTree dt, Map<Long, Integer> sessions,
PlayBackListener listener) throws IOException {
// 经过快照还原后，dt.lastProcessedZxid即是快照文件中存放的最大事务id
// 从事务日志中接着该事务id接着还原
TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
long highestZxid = dt.lastProcessedZxid;
TxnHeader hdr;
try {
while (true) {
// iterator points to
// the first valid txn when initialized
hdr = itr.getHeader();
if (hdr == null) {
//empty logs
return dt.lastProcessedZxid;
}
// 更新目前处理过的最大事务id
if (hdr.getZxid() < highestZxid && highestZxid != 0) {
LOG.error("{}(highestZxid) > {}(next log) for type {}",
highestZxid, hdr.getZxid(), hdr.getType());
} else {
highestZxid = hdr.getZxid();
}
try {
// 处理事务
processTransaction(hdr,dt,sessions, itr.getTxn());
} catch(KeeperException.NoNodeException e) {
throw new IOException("Failed to process transaction type: " +
hdr.getType() + " error: " + e.getMessage(), e);
}
listener.onTxnLoaded(hdr, itr.getTxn());
if (!itr.next())
break;
}
} finally {
if (itr != null) {
itr.close();
}
}
return highestZxid;
}
版权声明：「DDKK.COM 弟弟快看，程序员编程资料站」本站文章，版权归原作者所有