zookeeper源码(05)数据存储

本文详细分析一下zookeeper的数据存储。
ZKDatabase

维护zookeeper服务器内存数据库，包括session、dataTree和committedlog数据，从磁盘读取日志和快照后启动。
关键字段

1. // 数据节点树
2. protected DataTree dataTree;
3. protected ConcurrentHashMap<Long, Integer> sessionsWithTimeouts;
4. protected FileTxnSnapLog snapLog; // 用于操作底层数据文件
5. // committedLog中第一条和最后一条数据的zxid
6. protected long minCommittedLog, maxCommittedLog;
7. // committedLog最大容量，默认500
8. public int commitLogCount;
9. // 维护最后提交的请求集，可用于快速follower同步
10. protected Queue<Proposal> committedLog = new ArrayDeque<>();
12. protected ReentrantReadWriteLock logLock = new ReentrantReadWriteLock();
13. private volatile boolean initialized = false;
15. // txnlog计数
16. private AtomicInteger txnCount = new AtomicInteger(0);

复制代码

构造方法

1. public ZKDatabase(FileTxnSnapLog snapLog) {
2.     dataTree = createDataTree();
3.     sessionsWithTimeouts = new ConcurrentHashMap<>();
4.     this.snapLog = snapLog;
6.     // 初始化snapshotSizeFactor默认0.33
7.     // 初始化commitLogCount默认500
8. }
10. public DataTree createDataTree() {
11.     return new DataTree();
12. }

复制代码

创建DataTree对象：创建/zookeeper/quota、/zookeeper/config节点，创建dataWatches和childWatches对象(使用WatchManager实现类)。
主要方法

1. // 返回committedLog集
2. public synchronized Collection<Proposal> getCommittedLog();
3. // 返回dataTree.lastProcessedZxid的值
4. public long getDataTreeLastProcessedZxid();
5. // 返回dataTree.getSessions()集
6. public Collection<Long> getSessions();
7. // 返回sessionsWithTimeouts的size
8. public long getSessionCount();
9. // 从磁盘加载dataTree并把txnLog加载到committedLog中
10. public long loadDataBase() throws IOException;
11. // 从磁盘加载txnLog到committedLog中
12. public long fastForwardDataBase() throws IOException;
13. // 使用addCommittedProposal方法添加committedLog
14. private void addCommittedProposal(TxnHeader hdr, Record txn, TxnDigest digest);
15. // 添加committedLog
16. public void addCommittedProposal(Request request);
17. // 从txnLog加载Proposal
18. public Iterator<Proposal> getProposalsFromTxnLog(long startZxid, long sizeLimit);
19. // 使用dataTree.removeCnxn(cnxn)
20. public void removeCnxn(ServerCnxn cnxn);
21. // 使用dataTree.killSession(sessionId, zxid)
22. public void killSession(long sessionId, long zxid);
23. // 使用dataTree.dumpEphemerals(pwriter)
24. public void dumpEphemerals(PrintWriter pwriter);
25. // 使用dataTree.getEphemerals()
26. public Map<Long, Set<String>> getEphemerals();
27. // 使用dataTree.getNodeCount()
28. public int getNodeCount();
29. // 使用dataTree.getEphemerals(sessionId)
30. public Set<String> getEphemerals(long sessionId);
31. // 给dataTree.lastProcessedZxid赋值
32. public void setlastProcessedZxid(long zxid);
33. // 使用dataTree.processTxn(hdr, txn, digest)
34. public ProcessTxnResult processTxn(TxnHeader hdr, Record txn, TxnDigest digest);
35. // 使用dataTree.statNode(path, serverCnxn)
36. public Stat statNode(String path, ServerCnxn serverCnxn) throws KeeperException.NoNodeException;
37. // 使用dataTree.getNode(path)
38. public DataNode getNode(String path);
39. // 使用dataTree.getData(path, stat, watcher)
40. public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException;
41. // 使用dataTree.setWatches方法实现
42. public void setWatches(long relativeZxid, List<String> dataWatches,
43.                        List<String> existWatches, List<String> childWatches,
44.                        List<String> persistentWatches, List<String> persistentRecursiveWatches,
45.                        Watcher watcher);
46. // 使用dataTree.addWatch(basePath, watcher, mode)
47. public void addWatch(String basePath, Watcher watcher, int mode);
48. // 使用dataTree.getChildren(path, stat, watcher)
49. public List<String> getChildren(
50.     String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException;
51. // 使用dataTree.getAllChildrenNumber(path)
52. public int getAllChildrenNumber(String path) throws KeeperException.NoNodeException;
53. // Truncate the ZKDatabase to the specified zxid
54. public boolean truncateLog(long zxid) throws IOException;
55. // Deserialize a snapshot from an input archive
56. public void deserializeSnapshot(InputArchive ia) throws IOException;
57. // Deserialize a snapshot that contains FileHeader from an input archive
58. // It is used by the admin restore command
59. public void deserializeSnapshot(final InputArchive ia, final CheckedInputStream is) throws IOException;
60. // Serialize the snapshot
61. public void serializeSnapshot(OutputArchive oa) throws IOException, InterruptedException;
62. // 使用snapLog.append(si)保存数据，txnCount++
63. public boolean append(Request si) throws IOException;
64. // 使用snapLog.rollLog()滚动底层txnLog
65. public void rollLog() throws IOException;
66. // 使用snapLog.commit()提交底层txnLog
67. public void commit() throws IOException;
68. // 初始化/zookeeper/config数据，集群启动时已介绍
69. public synchronized void initConfigInZKDatabase(QuorumVerifier qv);
70. // 使用dataTree.containsWatcher(path, type, watcher)
71. public boolean containsWatcher(String path, WatcherType type, Watcher watcher);
72. // 使用dataTree.removeWatch(path, type, watcher)
73. public boolean removeWatch(String path, WatcherType type, Watcher watcher);

复制代码

loadDataBase方法

从磁盘加载dataTree并把txnLog加载到committedLog中：

1. public long loadDataBase() throws IOException {
2.     long startTime = Time.currentElapsedTime();
3.     // 1. 从snapshot加载dataTree
4.     // 2. 使用fastForwardFromEdits方法从txnLog加载dataTree和committedlog
5.     long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
6.     initialized = true;
7.     // 略
8.     return zxid;
9. }

复制代码

fastForwardDataBase方法

从txnLog加载dataTree和committedlog集：

1. public long fastForwardDataBase() throws IOException {
2.     // 会通过commitProposalPlaybackListener调用addCommittedProposal添加committedlog
3.     long zxid = snapLog.fastForwardFromEdits(
4.         dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
5.     initialized = true;
6.     return zxid;
7. }

复制代码

addCommittedProposal方法

1. private void addCommittedProposal(TxnHeader hdr, Record txn, TxnDigest digest) {
2.     Request r = new Request(0, hdr.getCxid(), hdr.getType(), hdr, txn, hdr.getZxid());
3.     r.setTxnDigest(digest);
4.     addCommittedProposal(r);
5. }
7. public void addCommittedProposal(Request request) {
8.     WriteLock wl = logLock.writeLock();
9.     try {
10.         wl.lock();
11.         if (committedLog.size() > commitLogCount) {
12.             committedLog.remove();
13.             minCommittedLog = committedLog.peek().packet.getZxid();
14.         }
15.         if (committedLog.isEmpty()) {
16.             minCommittedLog = request.zxid;
17.             maxCommittedLog = request.zxid;
18.         }
19.         byte[] data = request.getSerializeData();
20.         QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
21.         Proposal p = new Proposal();
22.         p.packet = pp;
23.         p.request = request;
24.         committedLog.add(p);
25.         maxCommittedLog = p.packet.getZxid();
26.     } finally {
27.         wl.unlock();
28.     }
29. }

复制代码

getProposalsFromTxnLog方法

从txnlog获取Proposal，只填充packet字段：

1. public Iterator<Proposal> getProposalsFromTxnLog(long startZxid, long sizeLimit) {
2.     if (sizeLimit < 0) {
3.         return TxnLogProposalIterator.EMPTY_ITERATOR;
4.     }
6.     TxnIterator itr = null;
7.     try {
8.         // 从txnLog文件读取数据
9.         // 底层通过FileTxnIterator类读取文件流实现
10.         itr = snapLog.readTxnLog(startZxid, false);
12.         // If we cannot guarantee that this is strictly the starting txn
13.         // after a given zxid, we should fail.
14.         if ((itr.getHeader() != null) && (itr.getHeader().getZxid() > startZxid)) {
15.             itr.close();
16.             return TxnLogProposalIterator.EMPTY_ITERATOR;
17.         }
19.         if (sizeLimit > 0) {
20.             long txnSize = itr.getStorageSize();
21.             if (txnSize > sizeLimit) {
22.                 itr.close();
23.                 return TxnLogProposalIterator.EMPTY_ITERATOR;
24.             }
25.         }
26.     } catch (IOException e) {
27.         itr.close();
28.         return TxnLogProposalIterator.EMPTY_ITERATOR;
29.     }
30.     return new TxnLogProposalIterator(itr);
31. }

复制代码

truncateLog方法

把txnlog数据truncate到指定的zxid位置，然后重新加载DataTree数据：

1. public boolean truncateLog(long zxid) throws IOException {
2.     clear();
4.     // truncate the log
5.     boolean truncated = snapLog.truncateLog(zxid);
7.     if (!truncated) {
8.         return false;
9.     }
11.     loadDataBase();
12.     return true;
13. }

复制代码

deserializeSnapshot方法

1. public void deserializeSnapshot(InputArchive ia) throws IOException {
2.     clear();
3.     SerializeUtils.deserializeSnapshot(getDataTree(), ia, getSessionWithTimeOuts());
4.     initialized = true;
5. }
7. public void deserializeSnapshot(final InputArchive ia, final CheckedInputStream is) throws IOException {
8.     clear();
10.     // deserialize data tree
11.     final DataTree dataTree = getDataTree();
12.     FileSnap.deserialize(dataTree, getSessionWithTimeOuts(), ia);
13.     SnapStream.checkSealIntegrity(is, ia);
15.     // deserialize digest and check integrity
16.     if (dataTree.deserializeZxidDigest(ia, 0)) {
17.         SnapStream.checkSealIntegrity(is, ia);
18.     }
20.     // deserialize lastProcessedZxid and check integrity
21.     if (dataTree.deserializeLastProcessedZxid(ia)) {
22.         SnapStream.checkSealIntegrity(is, ia);
23.     }
25.     // compare the digest to find inconsistency
26.     if (dataTree.getDigestFromLoadedSnapshot() != null) {
27.         dataTree.compareSnapshotDigests(dataTree.lastProcessedZxid);
28.     }
30.     initialized = true;
31. }

复制代码

serializeSnapshot方法

1. public void serializeSnapshot(OutputArchive oa) throws IOException, InterruptedException {
2.     SerializeUtils.serializeSnapshot(getDataTree(), oa, getSessionWithTimeOuts());
3. }

复制代码

DataTree

维护树状结构，没有任何网络或客户端连接代码，因此可以以独立的方式进行测试。
维护两个并行的数据结构：一个从完整路径映射到DataNodes的哈希表和一个DataNodes树，对路径的所有访问都是通过哈希表进行的，只有在序列化到磁盘时才遍历DataNodes树。
关键字段

1. // This map provides a fast lookup to the data nodes
2. private final NodeHashMap nodes;
3. // Watcher
4. private IWatchManager dataWatches;
5. private IWatchManager childWatches;
6. // cached total size of paths and data for all DataNodes
7. private final AtomicLong nodeDataSize = new AtomicLong(0);
8. // This hashtable lists the paths of the ephemeral nodes of a session
9. private final Map<Long, HashSet<String>> ephemerals = new ConcurrentHashMap<>();
10. // This set contains the paths of all container nodes
11. private final Set<String> containers = Collections.newSetFromMap(new ConcurrentHashMap<>());
12. // This set contains the paths of all ttl nodes
13. private final Set<String> ttls = Collections.newSetFromMap(new ConcurrentHashMap<>());
14. // This is a pointer to the root of the DataTree
15. private DataNode root = new DataNode(new byte[0], -1L, new StatPersisted());
16. // create a /zookeeper filesystem that is the proc filesystem of zookeeper
17. private final DataNode procDataNode = new DataNode(new byte[0], -1L, new StatPersisted());
18. // create a /zookeeper/quota node for maintaining quota properties for zookeeper
19. private final DataNode quotaDataNode = new DataNode(new byte[0], -1L, new StatPersisted());
20. // 最新被处理的zxid
21. public volatile long lastProcessedZxid = 0;

复制代码

• NodeHashMap - NodeHashMapImpl实现类使用ConcurrentHashMap保存path -> DataNode数据
  
• IWatchManager和Watcher - 监听器管理
  
• DataNode - 封装树节点信息，包括data、children、stat等
  

构造方法

1. DataTree(DigestCalculator digestCalculator) {
2.     this.digestCalculator = digestCalculator;
3.     nodes = new NodeHashMapImpl(digestCalculator);
5.     // rather than fight it, let root have an alias
6.     nodes.put("", root); // "" -> root
7.     nodes.putWithoutDigest(rootZookeeper, root); // "/" -> root
9.     // add the proc node and quota node
10.     root.addChild(procChildZookeeper); // 添加zookeeper子节点
11.     nodes.put(procZookeeper, procDataNode); // "/zookeeper" -> procDataNode
13.     procDataNode.addChild(quotaChildZookeeper); // 添加quota子节点
14.     nodes.put(quotaZookeeper, quotaDataNode); // "/zookeeper/quota" -> quotaDataNode
16.     addConfigNode(); // 添加/zookeeper/config节点
18.     nodeDataSize.set(approximateDataSize());
19.     try {
20.         // 使用WatchManager实现类
21.         dataWatches = WatchManagerFactory.createWatchManager();
22.         childWatches = WatchManagerFactory.createWatchManager();
23.     } catch (Exception e) {}
24. }
26. public void addConfigNode() {
27.     DataNode zookeeperZnode = nodes.get(procZookeeper); // 找到/zookeeper节点
28.     if (zookeeperZnode != null) {
29.         zookeeperZnode.addChild(configChildZookeeper); // 添加config子节点
30.     }
32.     nodes.put(configZookeeper, new DataNode(new byte[0], -1L, new StatPersisted()));
33.     try {
34.         // Reconfig node is access controlled by default (ZOOKEEPER-2014).
35.         setACL(configZookeeper, ZooDefs.Ids.READ_ACL_UNSAFE, -1);
36.     } catch (NoNodeException e) {}
37. }

复制代码

主要方法

1. // Add a new node to the DataTree
2. public void createNode(final String path, byte[] data, List<ACL> acl, long ephemeralOwner,
3.                        int parentCVersion, long zxid, long time, Stat outputStat);
4. // Remove path from the DataTree
5. public void deleteNode(String path, long zxid);
6. // 为节点设置数据
7. public Stat setData(String path, byte[] data, int version, long zxid, long time);
8. // 1. 获取path的data
9. // 2. 如果watcher不为null则addWatch
10. public byte[] getData(String path, Stat stat, Watcher watcher);
11. // 使用node.copyStat(stat)保存stat数据
12. public Stat statNode(String path, Watcher watcher);
14. // 1. copyStat到stat中
15. // 2. addWatch
16. // 3. getChildren
17. public List<String> getChildren(String path, Stat stat, Watcher watcher);
19. // 设置、获取权限
20. public Stat setACL(String path, List<ACL> acl, int version);
21. public List<ACL> getACL(String path, Stat stat);
22. public List<ACL> getACL(DataNode node);
23. // 添加Watcher
24. public void addWatch(String basePath, Watcher watcher, int mode);
26. // 处理事务请求
27. public ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn);
29. // 杀会话，使用deleteNodes删除paths2DeleteLocal和paths2DeleteInTxn集
30. void killSession(
31.     long session, long zxid, Set<String> paths2DeleteLocal, List<String> paths2DeleteInTxn);
32. // 遍历paths2Delete调用deleteNode方法删除节点
33. void deleteNodes(long session, long zxid, Iterable<String> paths2Delete);
34. // 递归方式获取path下面的总节点数和总字节数
35. private void getCounts(String path, Counts counts);
37. // 序列化
38. void serializeNode(OutputArchive oa, StringBuilder path);
39. public void serializeNodeData(OutputArchive oa, String path, DataNode node);
40. public void serializeAcls(OutputArchive oa);
41. public void serializeNodes(OutputArchive oa);
42. public void serialize(OutputArchive oa, String tag);
43. // 反序列化
44. public void deserialize(InputArchive ia, String tag);
45. // 从dataWatches和childWatches移除watcher
46. public void removeCnxn(Watcher watcher);
47. // 触发或addWatch
48. public void setWatches(long relativeZxid, List<String> dataWatches,
49.                        List<String> existWatches, List<String> childWatches,
50.                        List<String> persistentWatches, List<String> persistentRecursiveWatches,
51.                        Watcher watcher);
52. // 为path设置新的cversion和zxid
53. public void setCversionPzxid(String path, int newCversion, long zxid);
54. // Add the digest to the historical list, and update the latest zxid digest
55. private void logZxidDigest(long zxid, long digest);
56. // 序列化、反序列化lastProcessedZxidDigest
57. public boolean serializeZxidDigest(OutputArchive oa);
58. public boolean deserializeZxidDigest(InputArchive ia, long startZxidOfSnapshot);
59. // 序列化、反序列化lastProcessedZxid
60. public boolean serializeLastProcessedZxid(final OutputArchive oa);
61. public boolean deserializeLastProcessedZxid(final InputArchive ia);
62. // Compares the actual tree's digest with that in the snapshot.
63. // Resets digestFromLoadedSnapshot after comparison.
64. public void compareSnapshotDigests(long zxid);
65. // Compares the digest of the tree with the digest present in transaction digest.
66. // If there is any error, logs and alerts the watchers.
67. public boolean compareDigest(TxnHeader header, Record txn, TxnDigest digest);

复制代码

createNode方法

processTxn中会使用该方法创建节点：

1. public void createNode(final String path, byte[] data, List<ACL> acl,
2.                        long ephemeralOwner, int parentCVersion, long zxid,
3.                        long time, Stat outputStat) throws NoNodeException, NodeExistsException {
4.     int lastSlash = path.lastIndexOf('/');
5.     String parentName = path.substring(0, lastSlash);
6.     String childName = path.substring(lastSlash + 1);
7.     StatPersisted stat = createStat(zxid, time, ephemeralOwner); // Create a node stat
8.     DataNode parent = nodes.get(parentName); // 父节点需要存在
9.     synchronized (parent) {
10.         Long acls = aclCache.convertAcls(acl);
12.         Set<String> children = parent.getChildren(); // path节点不能存在
14.         nodes.preChange(parentName, parent); // 执行removeDigest
15.         if (parentCVersion == -1) {
16.             parentCVersion = parent.stat.getCversion();
17.             parentCVersion++; // childVersion递增
18.         }
20.         if (parentCVersion > parent.stat.getCversion()) {
21.             parent.stat.setCversion(parentCVersion); // 父节点的childVersion
22.             parent.stat.setPzxid(zxid); // 父节点processZxid
23.         }
24.         DataNode child = new DataNode(data, acls, stat);
25.         parent.addChild(childName); // 添加节点
26.         nodes.postChange(parentName, parent);
27.         nodeDataSize.addAndGet(getNodeSize(path, child.data));
28.         nodes.put(path, child); // 维护NodeHashMap
29.         EphemeralType ephemeralType = EphemeralType.get(ephemeralOwner);// 通常是VOID|NORMAL
30.         if (ephemeralType == EphemeralType.CONTAINER) {
31.             containers.add(path);
32.         } else if (ephemeralType == EphemeralType.TTL) {
33.             ttls.add(path);
34.         } else if (ephemeralOwner != 0) {
35.             // 维护临时节点
36.             HashSet<String> list = ephemerals.computeIfAbsent(ephemeralOwner, k -> new HashSet<>());
37.             synchronized (list) {
38.                 list.add(path);
39.             }
40.         }
41.         if (outputStat != null) {
42.             child.copyStat(outputStat); // 把权限保存到outputStat中
43.         }
44.     }
46.     // 略
48.     // 触发监听器
49.     dataWatches.triggerWatch(path, Event.EventType.NodeCreated, zxid);
50.     childWatches.triggerWatch(
51.         parentName.equals("") ? "/" : parentName, Event.EventType.NodeChildrenChanged, zxid);
52. }

复制代码

deleteNode方法

1. public void deleteNode(String path, long zxid) throws NoNodeException {
2.     int lastSlash = path.lastIndexOf('/');
3.     String parentName = path.substring(0, lastSlash);
4.     String childName = path.substring(lastSlash + 1);
6.     DataNode parent = nodes.get(parentName); // 父节点要存在
7.     synchronized (parent) {
8.         nodes.preChange(parentName, parent);
9.         parent.removeChild(childName); // 移除子节点
10.         if (zxid > parent.stat.getPzxid()) {
11.             parent.stat.setPzxid(zxid);
12.         }
13.         nodes.postChange(parentName, parent);
14.     }
16.     DataNode node = nodes.get(path); // 节点要存在
17.     nodes.remove(path); // 从NodeHashMap移除
18.     synchronized (node) {
19.         aclCache.removeUsage(node.acl);
20.         nodeDataSize.addAndGet(-getNodeSize(path, node.data));
21.     }
23.     synchronized (parent) {
24.         long owner = node.stat.getEphemeralOwner();
25.         EphemeralType ephemeralType = EphemeralType.get(owner);
26.         if (ephemeralType == EphemeralType.CONTAINER) {
27.             containers.remove(path);
28.         } else if (ephemeralType == EphemeralType.TTL) {
29.             ttls.remove(path);
30.         } else if (owner != 0) { // 移除临时节点
31.             Set<String> nodes = ephemerals.get(owner);
32.             if (nodes != null) {
33.                 synchronized (nodes) {
34.                     nodes.remove(path);
35.                 }
36.             }
37.         }
38.     }
40.     // 略
42.     // 触发监听器
43.     WatcherOrBitSet processed = dataWatches.triggerWatch(path, EventType.NodeDeleted, zxid);
44.     childWatches.triggerWatch(path, EventType.NodeDeleted, zxid, processed);
45.     childWatches.triggerWatch(
46.         "".equals(parentName) ? "/" : parentName, EventType.NodeChildrenChanged, zxid);
47. }

复制代码

setData方法

1. public Stat setData(String path, byte[] data, int version, long zxid, long time) throws NoNodeException {
2.     Stat s = new Stat();
3.     DataNode n = nodes.get(path); // 节点要存在
4.     byte[] lastData;
5.     synchronized (n) {
6.         lastData = n.data;
7.         nodes.preChange(path, n);
8.         n.data = data; // data赋值
9.         n.stat.setMtime(time); // 修改时间
10.         n.stat.setMzxid(zxid); // 修改的zxid
11.         n.stat.setVersion(version); // 版本
12.         n.copyStat(s); // 保存stat
13.         nodes.postChange(path, n);
14.     }
16.     // 略
17.     // 触发监听器
18.     dataWatches.triggerWatch(path, EventType.NodeDataChanged, zxid);
19.     return s;
20. }

复制代码

setAcl等acl方法

1. public Stat setACL(String path, List<ACL> acl, int version) throws NoNodeException {
2.     DataNode n = nodes.get(path);
3.     synchronized (n) {
4.         Stat stat = new Stat();
5.         aclCache.removeUsage(n.acl);
6.         nodes.preChange(path, n);
7.         n.stat.setAversion(version); // access时间
8.         n.acl = aclCache.convertAcls(acl); // 设置权限
9.         n.copyStat(stat);
10.         nodes.postChange(path, n);
11.         return stat;
12.     }
13. }
15. public List<ACL> getACL(String path, Stat stat) throws NoNodeException {
16.     DataNode n = nodes.get(path);
17.     synchronized (n) {
18.         if (stat != null) {
19.             n.copyStat(stat);
20.         }
21.         return new ArrayList<>(aclCache.convertLong(n.acl));
22.     }
23. }
25. public List<ACL> getACL(DataNode node) {
26.     synchronized (node) {
27.         return aclCache.convertLong(node.acl);
28.     }
29. }

复制代码

addWatch方法

1. public void addWatch(String basePath, Watcher watcher, int mode) {
2.     WatcherMode watcherMode = WatcherMode.fromZooDef(mode); // PERSISTENT_RECURSIVE or PERSISTENT
3.     dataWatches.addWatch(basePath, watcher, watcherMode); // 只给节点添加Watcher
4.     if (watcherMode != WatcherMode.PERSISTENT_RECURSIVE) {
5.         childWatches.addWatch(basePath, watcher, watcherMode); // 递归添加Watcher
6.     }
7. }

复制代码

processTxn方法

1. public ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn) {
2.     ProcessTxnResult rc = new ProcessTxnResult();
4.     try {
5.         rc.clientId = header.getClientId();
6.         rc.cxid = header.getCxid();
7.         rc.zxid = header.getZxid();
8.         rc.type = header.getType();
9.         rc.err = 0;
10.         rc.multiResult = null;
11.         switch (header.getType()) {
12.         case OpCode.create: // 创建节点
13.             CreateTxn createTxn = (CreateTxn) txn;
14.             rc.path = createTxn.getPath();
15.             createNode(createTxn.getPath(), createTxn.getData(), createTxn.getAcl(),
16.                 createTxn.getEphemeral() ? header.getClientId() : 0, createTxn.getParentCVersion(),
17.                 header.getZxid(), header.getTime(), null);
18.             break;
19.         case OpCode.create2: // 创建节点并保存stat
20.             CreateTxn create2Txn = (CreateTxn) txn;
21.             rc.path = create2Txn.getPath();
22.             Stat stat = new Stat();
23.             createNode(create2Txn.getPath(), create2Txn.getData(), create2Txn.getAcl(),
24.                 create2Txn.getEphemeral() ? header.getClientId() : 0, create2Txn.getParentCVersion(),
25.                 header.getZxid(), header.getTime(), stat);
26.             rc.stat = stat;
27.             break;
28.         case OpCode.createTTL:
29.             CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
30.             rc.path = createTtlTxn.getPath();
31.             stat = new Stat();
32.             createNode(createTtlTxn.getPath(), createTtlTxn.getData(), createTtlTxn.getAcl(),
33.                 EphemeralType.TTL.toEphemeralOwner(createTtlTxn.getTtl()), // ttl
34.                 createTtlTxn.getParentCVersion(), header.getZxid(), header.getTime(), stat);
35.             rc.stat = stat;
36.             break;
37.         case OpCode.createContainer:
38.             CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
39.             rc.path = createContainerTxn.getPath();
40.             stat = new Stat();
41.             createNode(createContainerTxn.getPath(), createContainerTxn.getData(),
42.                 createContainerTxn.getAcl(), EphemeralType.CONTAINER_EPHEMERAL_OWNER,
43.                 createContainerTxn.getParentCVersion(), header.getZxid(), header.getTime(), stat);
44.             rc.stat = stat;
45.             break;
46.         case OpCode.delete:
47.         case OpCode.deleteContainer:
48.             DeleteTxn deleteTxn = (DeleteTxn) txn;
49.             rc.path = deleteTxn.getPath();
50.             deleteNode(deleteTxn.getPath(), header.getZxid()); // 删除节点
51.             break;
52.         case OpCode.reconfig:
53.         case OpCode.setData: // 设置节点数据
54.             SetDataTxn setDataTxn = (SetDataTxn) txn;
55.             rc.path = setDataTxn.getPath();
56.             rc.stat = setData(setDataTxn.getPath(), setDataTxn.getData(), setDataTxn.getVersion(),
57.                 header.getZxid(), header.getTime());
58.             break;
59.         case OpCode.setACL: // 设置ACL
60.             SetACLTxn setACLTxn = (SetACLTxn) txn;
61.             rc.path = setACLTxn.getPath();
62.             rc.stat = setACL(setACLTxn.getPath(), setACLTxn.getAcl(), setACLTxn.getVersion());
63.             break;
64.         case OpCode.closeSession: // 关闭session
65.             long sessionId = header.getClientId();
66.             if (txn != null) {
67.                 killSession(sessionId, header.getZxid(), ephemerals.remove(sessionId),
68.                         ((CloseSessionTxn) txn).getPaths2Delete());
69.             } else {
70.                 killSession(sessionId, header.getZxid());
71.             }
72.             break;
73.         case OpCode.error:
74.             ErrorTxn errTxn = (ErrorTxn) txn;
75.             rc.err = errTxn.getErr();
76.             break;
77.         case OpCode.check:
78.             CheckVersionTxn checkTxn = (CheckVersionTxn) txn;
79.             rc.path = checkTxn.getPath();
80.             break;
81.         case OpCode.multi:
82.             // 遍历处理每一个Txn
83.             break;
84.         }
85.     } catch (KeeperException e) {
86.         rc.err = e.code().intValue();
87.     } catch (IOException e) {}
89.     //
90.     if (header.getType() == OpCode.create && rc.err == Code.NODEEXISTS.intValue()) {
91.         int lastSlash = rc.path.lastIndexOf('/');
92.         String parentName = rc.path.substring(0, lastSlash);
93.         CreateTxn cTxn = (CreateTxn) txn;
94.         try {
95.             setCversionPzxid(parentName, cTxn.getParentCVersion(), header.getZxid());
96.         } catch (NoNodeException e) {
97.             rc.err = e.code().intValue();
98.         }
99.     }
100.     //
101.     if (!isSubTxn) {
102.         if (rc.zxid > lastProcessedZxid) {
103.             lastProcessedZxid = rc.zxid; // 设置最新lastProcessedZxid
104.         }
106.         // 略
107.     }
109.     return rc;
110. }

复制代码

serialize相关方法

1. void serializeNode(OutputArchive oa, StringBuilder path) throws IOException {
2.     String pathString = path.toString();
3.     DataNode node = getNode(pathString); // 查找节点
4.     String[] children;
5.     DataNode nodeCopy;
6.     synchronized (node) {
7.         StatPersisted statCopy = new StatPersisted();
8.         copyStatPersisted(node.stat, statCopy);
9.         // we do not need to make a copy of node.data because the contents are never changed
10.         nodeCopy = new DataNode(node.data, node.acl, statCopy);
11.         children = node.getChildren().toArray(new String[0]);
12.     }
13.     serializeNodeData(oa, pathString, nodeCopy); // 把节点写入到oa中
14.     path.append('/');
15.     int off = path.length();
16.     // 遍历子节点，将子节点写入oa中
17.     for (String child : children) {
18.         path.delete(off, Integer.MAX_VALUE);
19.         path.append(child);
20.         serializeNode(oa, path);
21.     }
22. }
24. // visible for test
25. public void serializeNodeData(OutputArchive oa, String path, DataNode node) throws IOException {
26.     oa.writeString(path, "path");
27.     oa.writeRecord(node, "node");
28. }
30. public void serializeAcls(OutputArchive oa) throws IOException {
31.     aclCache.serialize(oa);
32. }
34. // 序列化整个NodeHashMap对象
35. public void serializeNodes(OutputArchive oa) throws IOException {
36.     serializeNode(oa, new StringBuilder());
37.     // / marks end of stream
38.     // we need to check if clear had been called in between the snapshot.
39.     if (root != null) {
40.         oa.writeString("/", "path");
41.     }
42. }
44. // 完整序列化
45. public void serialize(OutputArchive oa, String tag) throws IOException {
46.     serializeAcls(oa);
47.     serializeNodes(oa);
48. }

复制代码

deserialize相关方法

1. public void deserialize(InputArchive ia, String tag) throws IOException {
2.     aclCache.deserialize(ia);
3.     nodes.clear();
4.     pTrie.clear();
5.     nodeDataSize.set(0);
6.     String path = ia.readString("path");
7.     while (!"/".equals(path)) {
8.         DataNode node = new DataNode();
9.         ia.readRecord(node, "node");
10.         nodes.put(path, node);
11.         synchronized (node) {
12.             aclCache.addUsage(node.acl);
13.         }
14.         int lastSlash = path.lastIndexOf('/');
15.         if (lastSlash == -1) {
16.             root = node;
17.         } else {
18.             String parentPath = path.substring(0, lastSlash);
19.             DataNode parent = nodes.get(parentPath);
20.             if (parent == null) {
21.                 throw new IOException(
22.                         "Invalid Datatree, unable to find parent " + parentPath + " of path " + path);
23.             }
24.             parent.addChild(path.substring(lastSlash + 1));
25.             long owner = node.stat.getEphemeralOwner();
26.             EphemeralType ephemeralType = EphemeralType.get(owner);
27.             if (ephemeralType == EphemeralType.CONTAINER) {
28.                 containers.add(path);
29.             } else if (ephemeralType == EphemeralType.TTL) {
30.                 ttls.add(path);
31.             } else if (owner != 0) {
32.                 HashSet<String> list = ephemerals.computeIfAbsent(owner, k -> new HashSet<>());
33.                 list.add(path);
34.             }
35.         }
36.         path = ia.readString("path");
37.     }
38.     // have counted digest for root node with "", ignore here to avoid counting twice for root node
39.     nodes.putWithoutDigest("/", root);
41.     nodeDataSize.set(approximateDataSize());
43.     // we are done with deserializing the datatree update the quotas - create path trie
44.     // and also update the stat nodes
45.     setupQuota();
47.     aclCache.purgeUnused();
48. }

复制代码

FileTxnSnapLog

操作TxnLog和SnapShot的入口类。
构造方法会创建dataDir和snapDir目录，判断数据目录可写，创建txnLog和snapLog对象访问数据文件。
主要方法

1. // 从snapshots和transaction logs加载数据库
2. public long restore(DataTree dt, Map<Long, Integer> sessions, PlayBackListener listener);
4. // fast forward the server database to have the latest transactions in it
5. // This is the same as restore, but only reads from the transaction logs and not restores from a snapshot
6. public long fastForwardFromEdits(DataTree dt, Map<Long, Integer> sessions, PlayBackListener listener);
8. // 使用txnLog.read(zxid, fastForward)方法从指定zxid加载TxnIterator
9. public TxnIterator readTxnLog(long zxid, boolean fastForward);
11. // process the transaction on the datatree
12. public void processTransaction(TxnHeader hdr, DataTree dt, Map<Long, Integer> sessions, Record txn);
14. // 使用txnLog.getLastLoggedZxid()方法获取last logged zxid
15. public long getLastLoggedZxid();
17. // 把datatree和sessions保存到snapshot中
18. public File save(
19.     DataTree dataTree, ConcurrentHashMap<Long, Integer> sessionsWithTimeouts, boolean syncSnap);
21. // 把txnLog truncate到指定的zxid
22. public boolean truncateLog(long zxid);
24. // 使用snaplog.findMostRecentSnapshot()方法加载最近snapshot文件
25. public File findMostRecentSnapshot();
26. // 使用snaplog.findNRecentSnapshots(n)方法加载n个最近snapshot文件
27. public List<File> findNRecentSnapshots(int n);
28. // 使用snaplog.findNValidSnapshots(n)方法加载n个合法snapshot文件
29. public List<File> findNValidSnapshots(int n);
31. // 获取快照文件，可能包含比给定zxid更新的事务。
32. // 包括起始zxid大于给定zxid的日志，以及起始zxid小于给定zxid的最新事务日志。
33. // 后一个日志文件可能包含超出给定zxid的事务。
34. public File[] getSnapshotLogs(long zxid);
36. // 使用txnLog.append(si)追加数据
37. public boolean append(Request si);
38. // txnLog.commit()提交数据
39. public void commit();

复制代码

restore方法

• 从snapshot加载dataTree数据
  
• 从txnlog加载dataTree和committedlog数据
  
• 如果没有加载到dataTree数据，将空的dataTree数据保存到snapshot.0文件中
  

fastForwardFromEdits方法

从txnlog加载dataTree和committedlog数据。
processTransaction方法

1. public void processTransaction(TxnHeader hdr, DataTree dt, Map<Long, Integer> sessions,
2.         Record txn) throws KeeperException.NoNodeException {
3.     ProcessTxnResult rc;
4.     switch (hdr.getType()) {
5.     case OpCode.createSession:
6.         sessions.put(hdr.getClientId(), ((CreateSessionTxn) txn).getTimeOut());
7.         // give dataTree a chance to sync its lastProcessedZxid
8.         rc = dt.processTxn(hdr, txn);
9.         break;
10.     case OpCode.closeSession:
11.         sessions.remove(hdr.getClientId());
12.         rc = dt.processTxn(hdr, txn);
13.         break;
14.     default:
15.         rc = dt.processTxn(hdr, txn);
16.     }
17. }

复制代码

save方法

1. public File save(DataTree dataTree, ConcurrentHashMap<Long, Integer> sessionsWithTimeouts,
2.         boolean syncSnap) throws IOException {
3.     long lastZxid = dataTree.lastProcessedZxid;
4.     // 文件名snapshot.${lastZxid}
5.     File snapshotFile = new File(snapDir, Util.makeSnapshotName(lastZxid));
6.     try {
7.         snapLog.serialize(dataTree, sessionsWithTimeouts, snapshotFile, syncSnap);
8.         return snapshotFile;
9.     } catch (IOException e) {
10.         throw e;
11.     }
12. }

复制代码

truncateLog方法

1. public boolean truncateLog(long zxid) {
2.     try {
3.         // close the existing txnLog and snapLog
4.         close();
6.         // truncate it
7.         try (FileTxnLog truncLog = new FileTxnLog(dataDir)) {
8.             boolean truncated = truncLog.truncate(zxid);
10.             // re-open the txnLog and snapLog
11.             // I'd rather just close/reopen this object itself, however that
12.             // would have a big impact outside ZKDatabase as there are other
13.             // objects holding a reference to this object.
14.             txnLog = new FileTxnLog(dataDir);
15.             snapLog = new FileSnap(snapDir);
17.             return truncated;
18.         }
19.     } catch (IOException e) {
20.         return false;
21.     }
22. }

复制代码

TxnLog接口和FileTxnLog实现类

txnlog

使用文件保存所有的事务操作，客户端的写操作会先写入txnlog文件，在follower达到quorum状态后提交到dataTree中，在ZKDatabase启动阶段，如果txnlog的zxid大于snapshot的zxid时，会加载txnlog文件数据回放事务，提交到dataTree中。
TxnLog接口

Interface for reading transaction logs.

1. public interface TxnLog extends Closeable {
3.     // Setter for ServerStats to monitor fsync threshold exceed
4.     void setServerStats(ServerStats serverStats);
6.     // roll the current log being appended to
7.     void rollLog() throws IOException;
9.     // Append a request to the transaction log with a digset
10.     boolean append(Request request) throws IOException;
12.     // Start reading the transaction logs from a given zxid
13.     TxnIterator read(long zxid) throws IOException;
15.     // the last zxid of the logged transactions
16.     long getLastLoggedZxid() throws IOException;
18.     // truncate the log to get in sync with the leader
19.     boolean truncate(long zxid) throws IOException;
21.     // the dbid for this transaction log
22.     long getDbId() throws IOException;
24.     // commit the transaction and make sure they are persisted
25.     void commit() throws IOException;
27.     // return transaction log's elapsed sync time in milliseconds
28.     long getTxnLogSyncElapsedTime();
30.     void close() throws IOException;
31.     void setTotalLogSize(long size);
32.     long getTotalLogSize();
33. }

复制代码

FileTxnLog实现类

1. This class implements the TxnLog interface. It provides api's to access the txnlogs and add entries to it.
2. The format of a Transactional log is as follows:
4.    LogFile:
5.        FileHeader TxnList ZeroPad
7.    FileHeader: {
8.        magic 4bytes (ZKLG)
9.        version 4bytes
10.        dbid 8bytes
11.    }
13.    TxnList:
14.        Txn || Txn TxnList
16.    Txn:
17.        checksum Txnlen TxnHeader Record 0x42
19.    checksum: 8bytes Adler32 is currently used
20.      calculated across payload -- Txnlen, TxnHeader, Record and 0x42
22.    Txnlen:
23.        len 4bytes
25.    TxnHeader: {
26.        sessionid 8bytes
27.        cxid 4bytes
28.        zxid 8bytes
29.        time 8bytes
30.        type 4bytes
31.    }
33.    Record:
34.        See Jute definition file for details on the various record types
36.    ZeroPad:
37.        0 padded to EOF (filled during preallocation stage)

复制代码

FileTxnLog主要方法实现

1. public synchronized boolean append(Request request) throws IOException {
2.     TxnHeader hdr = request.getHdr();
3.     if (hdr == null) { // 不是事务请求
4.         return false;
5.     }
6.     if (hdr.getZxid() <= lastZxidSeen) {
7.         LOG.warn("...");
8.     } else {
9.         lastZxidSeen = hdr.getZxid();
10.     }
11.     if (logStream == null) {
12.         // 创建新log.${hdr.zxid}文件
13.         logFileWrite = new File(logDir, Util.makeLogName(hdr.getZxid()));
14.         fos = new FileOutputStream(logFileWrite);
15.         logStream = new BufferedOutputStream(fos);
16.         oa = BinaryOutputArchive.getArchive(logStream);
17.         FileHeader fhdr = new FileHeader(TXNLOG_MAGIC, VERSION, dbId); // 文件头
18.         long dataSize = oa.getDataSize();
19.         fhdr.serialize(oa, "fileheader"); // 写文件头
20.         logStream.flush();
21.         // 文件偏移量
22.         filePosition += oa.getDataSize() - dataSize;
23.         filePadding.setCurrentSize(filePosition);
24.         streamsToFlush.add(fos);
25.     }
26.     fileSize = filePadding.padFile(fos.getChannel(), filePosition);
27.     byte[] buf = request.getSerializeData();
28.     long dataSize = oa.getDataSize();
29.     Checksum crc = makeChecksumAlgorithm();
30.     crc.update(buf, 0, buf.length);
31.     oa.writeLong(crc.getValue(), "txnEntryCRC"); // checksum
32.     Util.writeTxnBytes(oa, buf); // 写len, hdr, txn, digest, 0x42
33.     unFlushedSize += oa.getDataSize() - dataSize; // 计算未flush字节数
34.     return true;
35. }
37. public long getLastLoggedZxid() {
38.     File[] files = getLogFiles(logDir.listFiles(), 0);
39.     long maxLog = files.length > 0 ?
40.         Util.getZxidFromName(files[files.length - 1].getName(), LOG_FILE_PREFIX) : -1;
41.     // 最新的log文件的后缀作为zxid
42.     long zxid = maxLog;
43.     // 从文件解析最新zxid
44.     try (FileTxnLog txn = new FileTxnLog(logDir); TxnIterator itr = txn.read(maxLog)) {
45.         while (true) {
46.             if (!itr.next()) {
47.                 break;
48.             }
49.             TxnHeader hdr = itr.getHeader();
50.             zxid = hdr.getZxid();
51.         }
52.     } catch (IOException e) {
53.     }
54.     return zxid;
55. }
57. public synchronized void rollLog() throws IOException {
58.     if (logStream != null) {
59.         this.logStream.flush(); // 把当前文件刷写出去
60.         prevLogsRunningTotal += getCurrentLogSize();
61.         this.logStream = null; // 重置相关变量，后续append时会创建新的文件
62.         oa = null;
63.         fileSize = 0;
64.         filePosition = 0;
65.         unFlushedSize = 0;
66.     }
67. }
69. public synchronized void commit() throws IOException {
70.     if (logStream != null) {
71.         logStream.flush(); // 刷写文件
72.         filePosition += unFlushedSize;
73.         // If we have written more than we have previously preallocated,
74.         // we should override the fileSize by filePosition.
75.         if (filePosition > fileSize) {
76.             fileSize = filePosition;
77.         }
78.         unFlushedSize = 0;
79.     }
80.     for (FileOutputStream log : streamsToFlush) {
81.         log.flush(); // 刷写文件
82.         if (forceSync) {
83.             long startSyncNS = System.nanoTime();
85.             FileChannel channel = log.getChannel();
86.             channel.force(false);
88.             // 略
89.         }
90.     }
91.     // 关闭文件流
92.     while (streamsToFlush.size() > 1) {
93.         streamsToFlush.poll().close();
94.     }
96.     // Roll the log file if we exceed the size limit
97.     if (txnLogSizeLimit > 0) { // 默认-1分支进不来
98.         long logSize = getCurrentLogSize();
99.         if (logSize > txnLogSizeLimit) {
100.             rollLog();
101.         }
102.     }
103. }
105. // FileTxnIterator封装logFile和输入流对象，可以按照协议从文件流读取txnLog数据
106. public TxnIterator read(long zxid) throws IOException {
107.     return read(zxid, true);
108. }
109. public TxnIterator read(long zxid, boolean fastForward) throws IOException {
110.     return new FileTxnIterator(logDir, zxid, fastForward);
111. }
113. // 将log文件truncate到指定zxid位置
114. public boolean truncate(long zxid) throws IOException {
115.     try (FileTxnIterator itr = new FileTxnIterator(this.logDir, zxid)) {
116.         PositionInputStream input = itr.inputStream;
117.         if (input == null) {
118.             throw new IOException("No log files found to truncate");
119.         }
120.         long pos = input.getPosition();
121.         // now, truncate at the current position
122.         RandomAccessFile raf = new RandomAccessFile(itr.logFile, "rw");
123.         raf.setLength(pos);
124.         raf.close(); // 把最小的文件truncate到指定zxid位置
125.         while (itr.goToNextLog()) { // 删除所有>zxid的log文件
126.             if (!itr.logFile.delete()) {
127.             }
128.         }
129.     }
130.     return true;
131. }
133. private static FileHeader readHeader(File file) throws IOException {
134.     InputStream is = null;
135.     try {
136.         is = new BufferedInputStream(new FileInputStream(file));
137.         InputArchive ia = BinaryInputArchive.getArchive(is);
138.         FileHeader hdr = new FileHeader();
139.         hdr.deserialize(ia, "fileheader"); // 反序列化
140.         return hdr;
141.     } finally {
142.         // is.close();
143.     }
144. }

复制代码

FileTxnIterator类

this class implements the txnlog iterator interface which is used for reading the transaction logs.
内部使用List保存着比指定zxid大或者含有指定zxid数据的log文件，初始化阶段会定位到参数zxid指定的位置，这样在后续访问时就可以从参数指定的zxid开始读取数据了。

1. public FileTxnIterator(File logDir, long zxid, boolean fastForward) throws IOException {
2.     this.logDir = logDir;
3.     this.zxid = zxid;
4.     init();
6.     if (fastForward && hdr != null) {
7.         while (hdr.getZxid() < zxid) { // 这里将数据移动到zxid位置
8.             if (!next()) {
9.                 break;
10.             }
11.         }
12.     }
13. }
15. void init() throws IOException {
16.     storedFiles = new ArrayList<>();
17.     // 倒序查找log文件
18.     List<File> files = Util.sortDataDir(
19.         FileTxnLog.getLogFiles(logDir.listFiles(), 0),
20.         LOG_FILE_PREFIX,
21.         false);
22.     for (File f : files) {
23.         if (Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX) >= zxid) {
24.             storedFiles.add(f);
25.         } else if (Util.getZxidFromName(f.getName(), LOG_FILE_PREFIX) < zxid) {
26.             // add the last logfile that is less than the zxid
27.             storedFiles.add(f);
28.             break;
29.         }
30.     }
31.     goToNextLog(); // 定位到下一个文件
32.     next(); // 定位到下一个log数据
33. }

复制代码

SnapShot接口和FileSnap实现类

SnapShot接口

snapshot interface for the persistence layer. implement this interface for implementing snapshots.

1. public interface SnapShot {
3.     // deserialize a data tree from the last valid snapshot and return the last zxid that was deserialized
4.     long deserialize(DataTree dt, Map<Long, Integer> sessions) throws IOException;
6.     // persist the datatree and the sessions into a persistence storage
7.     void serialize(DataTree dt, Map<Long, Integer> sessions, File name, boolean fsync) throws IOException;
9.     // find the most recent snapshot file
10.     File findMostRecentSnapshot() throws IOException;
12.     // get information of the last saved/restored snapshot
13.     SnapshotInfo getLastSnapshotInfo();
15.     // free resources from this snapshot immediately
16.     void close() throws IOException;
17. }

复制代码

FileSnap实现类

负责存储、序列化和反序列化正确的快照。并提供对快照的访问：

1. public long deserialize(DataTree dt, Map<Long, Integer> sessions) throws IOException {
2.     // 在snapDir下查找合法的快照文件，倒序，所以最新的在前面
3.     List<File> snapList = findNValidSnapshots(100);
4.     File snap = null;
5.     long snapZxid = -1;
6.     boolean foundValid = false;
7.     for (int i = 0, snapListSize = snapList.size(); i < snapListSize; i++) {
8.         snap = snapList.get(i);
9.         snapZxid = Util.getZxidFromName(snap.getName(), SNAPSHOT_FILE_PREFIX);
10.         try (CheckedInputStream snapIS = SnapStream.getInputStream(snap)) {
11.             InputArchive ia = BinaryInputArchive.getArchive(snapIS);
12.             deserialize(dt, sessions, ia); // 将数据反序列到dt
13.             SnapStream.checkSealIntegrity(snapIS, ia);
15.             // Deserializing the zxid digest from the input
16.             // stream and update the digestFromLoadedSnapshot.
17.             // 格式: zxid digestVersion digest
18.             if (dt.deserializeZxidDigest(ia, snapZxid)) {
19.                 SnapStream.checkSealIntegrity(snapIS, ia);
20.             }
22.             // deserialize lastProcessedZxid and check inconsistency
23.             // 读lastZxid字段得到
24.             if (dt.deserializeLastProcessedZxid(ia)) {
25.                 SnapStream.checkSealIntegrity(snapIS, ia);
26.             }
28.             foundValid = true;
29.             break;
30.         } catch (IOException e) {}
31.     }
32.     // 验证foundValid
33.     // 上次处理到的zxid
34.     dt.lastProcessedZxid = snapZxid;
35.     lastSnapshotInfo = new SnapshotInfo(dt.lastProcessedZxid, snap.lastModified() / 1000);
37.     // compare the digest if this is not a fuzzy snapshot, we want to compare and find inconsistent asap
38.     if (dt.getDigestFromLoadedSnapshot() != null) {
39.         dt.compareSnapshotDigests(dt.lastProcessedZxid);
40.     }
41.     return dt.lastProcessedZxid;
42. }
44. public static void deserialize(
45.         DataTree dt, Map<Long, Integer> sessions, InputArchive ia) throws IOException {
46.     FileHeader header = new FileHeader(); // magic, version, dbid
47.     header.deserialize(ia, "fileheader"); // 解析文件头并验证magic
48.     if (header.getMagic() != SNAP_MAGIC) {
49.         throw new IOException("mismatching magic headers");
50.     }
51.     // 反序列化
52.     // 会话:
53.     //   Count Session(s)
54.     //   Session {id, timeout}
55.     // 节点:
56.     //   AclCache PathNode(s)
57.     //   PathNode {path, node}
58.     //   node {data, acl, stat}
59.     SerializeUtils.deserializeSnapshot(dt, ia, sessions);
60. }
62. protected List<File> findNValidSnapshots(int n) {
63.     // 在snapDir下查找快照文件，倒序，最新的在前面
64.     List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
65.     int count = 0;
66.     List<File> list = new ArrayList<>();
67.     for (File f : files) {
68.         try {
69.             if (SnapStream.isValidSnapshot(f)) { // 验证文件合法
70.                 list.add(f);
71.                 count++;
72.                 if (count == n) {
73.                     break;
74.                 }
75.             }
76.         } catch (IOException e) {}
77.     }
78.     return list;
79. }
81. public List<File> findNRecentSnapshots(int n) throws IOException {
82.     // 在snapDir下查找快照文件，倒序，最新的在前面
83.     List<File> files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);
84.     int count = 0;
85.     List<File> list = new ArrayList<>();
86.     for (File f : files) {
87.         if (count == n) {
88.             break;
89.         }
90.         if (Util.getZxidFromName(f.getName(), SNAPSHOT_FILE_PREFIX) != -1) {
91.             count++;
92.             list.add(f);
93.         }
94.     }
95.     return list;
96. }
98. protected void serialize(
99.         DataTree dt, Map<Long, Integer> sessions, OutputArchive oa, FileHeader header) throws IOException {
100.     // 验证header!=null
101.     header.serialize(oa, "fileheader"); // 序列化文件头
102.     SerializeUtils.serializeSnapshot(dt, oa, sessions); // 序列化dataTree
103. }
105. public synchronized void serialize(
106.         DataTree dt, Map<Long, Integer> sessions, File snapShot, boolean fsync) throws IOException {
107.     if (!close) {
108.         try (CheckedOutputStream snapOS = SnapStream.getOutputStream(snapShot, fsync)) {
109.             OutputArchive oa = BinaryOutputArchive.getArchive(snapOS);
110.             FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
111.             serialize(dt, sessions, oa, header);
112.             SnapStream.sealStream(snapOS, oa);
114.             // 序列化digest
115.             if (dt.serializeZxidDigest(oa)) {
116.                 SnapStream.sealStream(snapOS, oa);
117.             }
119.             // 序列化lastProcessZxid
120.             if (dt.serializeLastProcessedZxid(oa)) {
121.                 SnapStream.sealStream(snapOS, oa);
122.             }
124.             lastSnapshotInfo = new SnapshotInfo(
125.                 Util.getZxidFromName(snapShot.getName(), SNAPSHOT_FILE_PREFIX),
126.                 snapShot.lastModified() / 1000);
127.         }
128.     } else {
129.         throw new IOException("FileSnap has already been closed");
130.     }
131. }

复制代码

DatadirCleanupManager

启动周期任务

清理过期文件，保留最新的snapRetainCount个snapshot文件和对应的txnlog文件，将其余过期的文件删除掉。
purgeInterval参数指定执行周期(小时)，默认0不开启清理功能。

1. public void start() {
2.     if (PurgeTaskStatus.STARTED == purgeTaskStatus) {
3.         return;
4.     }
5.     // Don't schedule the purge task with zero or negative purge interval.
6.     if (purgeInterval <= 0) {
7.         return;
8.     }
10.     timer = new Timer("PurgeTask", true);
11.     TimerTask task = new PurgeTask(dataLogDir, snapDir, snapRetainCount);
12.     timer.scheduleAtFixedRate(task, 0, TimeUnit.HOURS.toMillis(purgeInterval));
14.     purgeTaskStatus = PurgeTaskStatus.STARTED;
15. }

复制代码

ContainerManager

负责清理container节点，只能有leader管理。启动后，定期检查cversion>0且没有子级的container节点和ttl节点。尝试删除节点，删除的结果并不重要。如果提议失败或容器节点不为空，则没有任何危害。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！