ZooKeeper源码分析（一）---ZooKeeper接口介绍

一、Server角色

每个Server在工作过程中有三种状态：

　　① LOOKING：当前Server不知道leader是谁，正在搜寻。

　　② LEADING：当前Server即为选举出来的leader。

　　③ FOLLOWING：leader已经选举出来，当前Server与之同步。

QuorumPeer定义了server的类型，其中ServerState表示server类型，LeanerType表示当ServerState为FOLLOWING时是参与者还是观察者，前者称为follower，后者称为observer。

代码如下：

1. public class QuorumPeer extends Thread implements QuorumStats.Provider{
2. 　　　　public enum ServerState {
3. 　　　　　　LOOKING, FOLLOWING, LEADING, OBSERVING;
4. 　　　　}
5. 　　　　public enum LearnerType {
6. 　　　　　　PARTICIPANT, OBSERVER;
7. 　　　　}
8. }

二、Znode类型

CreateMode中定义了四种节点类型，分别对应：

　　PERSISTENT：永久节点

　　EPHEMERAL：临时节点

　　ERSISTENT_SEQUENTIAL：永久节点、序列化

　　EPHEMERAL_SEQUENTIAL：临时节点、序列化

代码如下：

1. public enum CreateMode {
2.    PERSISTENT (0, false, false),
3.    PERSISTENT_SEQUENTIAL (2, false, true),
4.    EPHEMERAL (1, true, false),
5.    EPHEMERAL_SEQUENTIAL (3, true, true);
6. }

三、Stat类

Stat类定义znode节点的元信息，主要成员变量如下：

1. public class Stat implements Record {
2.    private long czxid; // 创建时的zxid
3.    private long mzxid; // 最新修改的zxid
4.    private long ctime; // 创建时间 private
5.    long mtime; // 修改时间
6.    private int version; // 版本号，对应znode data
7.    private int cversion; // 版本号，对应子znode
8.    private int aversion; // 版本号，对应acl
9.    private long ephemeralOwner; // 临时节点对应的client session id，默认为0
10.    private int dataLength; // znode data长度
11.    private int numChildren; // 子znode个数
12.    private long pzxid; // 最新修改的zxid，貌似与mzxid重合了
13. }

注意：

Stat、StatPersisited、StatPersisitedV1三个类，其成员变量和逻辑基本一致，但StatPersisited类少了dataLength和numChildren属性，StatPersisitedV1类少了dataLength、 numChildren和pzxid属性，具体不同类用在什么地方待进一步分析。

三、DataNOde类

　　DataNode类记录了znode节点的所有信息，包括其父节点、子节点、数据内容、ACL信息、stat元数据等，主要成员变量如下：

1. public class DataNode implements Record {
2. 　　　　DataNode parent;
3. 　　　　byte data[];
4. 　　　　Long acl;
5. 　　　　public StatPersisted stat;
6. 　　　　private Set<String> children = null;
7. }

需要注意acl和children两个成员变量。

acl是Long型值,相当于aclkey,具体的ACL信息实际上保存在DataTree中的longKeyMap和aclKeyMap中,前者保存了整个目录树所有节点的ACL信息，类型是Map<Long, List<ACL>>可以根据aclkey获得某节点的ACL信息列表，后者则是该map的反向结构。

children 用于记录该节点的子节点列表信息,但保存的并不是DataNode类型,而是只保存了每个子节点路径名的最后部分,比如该节点为"/biglog ",子节点为"/biglog /test1"，那么children中保存"test1"这个相对路径，这么做的目的是：This should be synchronized on except deserializing (for speed up issues)。

三、DataTree类

DataTree类维护整个目录树结构，ConcurrentHashMap<String, DataNode> nodes保存了从完整路径到DataNode的hashtable，而DataNode中的Set<String> children保存了父子关系，即子节点的相对路径。通过某DataNode可以获知其任意子节点的相对路径，然后拼装成完整路径，再去DataTree的nodes中查找。所有对节点路径的访问都是通过nodes完成的。主要成员变量如下：

（1）DataTree类：

1.  /**
2. * This hashtable provides a fast lookup to the datanodes. The tree is the
3. * source of truth and is where all the locking occurs
4. */
5. private final ConcurrentHashMap<String, DataNode> nodes =new ConcurrentHashMap<String, DataNode>();
6. private final WatchManager dataWatches = new WatchManager();
7. private final WatchManager childWatches = new WatchManager();
8.  
9. private static final String rootZookeeper = "/"; //ZooKeeper树的根节点
10.  
11. private static final String procZookeeper = Quotas.procZookeeper;// ZooKeeper节点，作为管理和状态节点
12. private static final String procChildZookeeper = procZookeeper.substring(1);//存储根节点的子节点的字符串
13. //the zookeeper quota node that acts as the quota management node for zookeeper
14. private static final String quotaZookeeper = Quotas.quotaZookeeper; //ZooKeeper quota节点，作为ZooKeeper的配额管理节点
15. private static final String quotaChildZookeeper = quotaZookeeper.substring(procZookeeper.length() + 1); // 存储ZooKeeper节点的子节点字符串
16.  
17. private final PathTrie pTrie = new PathTrie(); //path trie跟踪在DataTree中的quota节点
18.  
19. //该hashtable列出了一个会话的临时节点路径
20. private final Map<Long, HashSet<String>> ephemerals =new ConcurrentHashMap<Long, HashSet<String>>();
21. //this is map from longs to acl's. It saves acl's being stored for each datanode.
22. public final Map<Long, List<ACL>> longKeyMap =new HashMap<Long, List<ACL>>();
23. //this a map from acls to long.
24. public final Map<List<ACL>, Long> aclKeyMap =new HashMap<List<ACL>, Long>();
25.  
26. //在DataTree中acls的数量
27. protected long aclIndex = 0;

（2）Quota类：

1. public class Quotas {
2.  
3.     // the zookeeper nodes that acts as the management and status node
4.     public static final String procZookeeper = "/zookeeper";
5.  
6.     // the zookeeper quota node that acts as the quota management node for zookeeper
7.     public static final String quotaZookeeper = "/zookeeper/quota";
8.  
9.     //the limit node that has the limit of a subtree
10.     public static final String limitNode = "zookeeper_limits";
11.  
12.     //the stat node that monitors the limit of a subtree.
13.     public static final String statNode = "zookeeper_stats";
14.  
15.     /**
16.      * return the quota path associated with this
17.      * prefix
18.      * @param path the actual path in zookeeper.
19.      * @return the limit quota path
20.      */
21.     public static String quotaPath(String path) {
22.         return quotaZookeeper + path +
23.         "/" + limitNode;
24.    }
25.  
26.     /**
27.      * return the stat quota path associated with this
28.      * prefix.
29.      * @param path the actual path in zookeeper
30.      * @return the stat quota path
31.      */
32.     public static String statPath(String path) {
33.         return quotaZookeeper + path + "/" +
34.         statNode;
35.     }
36. }

（2）StatsTrack类

1. //a class that represents the stats associated with quotas
2. public class StatsTrack {
3.     private int count;
4.     private long bytes;
5.     private String countStr = "count";
6.     private String byteStr = "bytes";
7.  
8.     public StatsTrack() {
9.         this(null);
10.     }
11.     /**
12.      * the stat string should be of the form count=int,bytes=long
13.      * if stats is called with null the count and bytes are initialized
14.      * to -1.
15.      * @param stats the stat string to be intialized with
16.      */
17.     public StatsTrack(String stats) {
18.         if (stats == null) {
19.             stats = "count=-1,bytes=-1";
20.         }
21.         String[] split = stats.split(",");
22.         if (split.length != 2) {
23.             throw new IllegalArgumentException("invalid string " + stats);
24.         }
25.         count = Integer.parseInt(split[0].split("=")[1]);
26.         bytes = Long.parseLong(split[1].split("=")[1]);
27.     }
28.  
29.  
30.     /**
31.      * get the count of nodes allowed as part of quota
32.      *
33.      * @return the count as part of this string
34.      */
35.     public int getCount() {
36.         return this.count;
37.     }
38.  
39.     /**
40.      * set the count for this stat tracker.
41.      *
42.      * @param count
43.      * the count to set with
44.      */
45.     public void setCount(int count) {
46.         this.count = count;
47.     }
48.  
49.     /**
50.      * get the count of bytes allowed as part of quota
51.      *
52.      * @return the bytes as part of this string
53.      */
54.     public long getBytes() {
55.         return this.bytes;
56.     }
57.  
58.     /**
59.      * set teh bytes for this stat tracker.
60.      *
61.      * @param bytes
62.      * the bytes to set with
63.      */
64.     public void setBytes(long bytes) {
65.         this.bytes = bytes;
66.     }
67.  
68.     @Override
69.     /*
70.      * returns the string that maps to this stat tracking.
71.      */
72.     public String toString() {
73.         return countStr + "=" + count + "," + byteStr + "=" + bytes;
74.     }
75. }

四、DataTree初始化

DataTree初始化要完成的工作，需要建立系统节点，包括/、/zookeeper、/zookeeper/quota三个znode。

下面先看一下DataNode的构造函数如下：

1. public DataNode(DataNode parent, byte data[], Long acl, StatPersisted stat) {
2.         this.parent = parent;
3.         this.data = data;
4.         this.acl = acl;
5.         this.stat = stat;
6. }

Datatree初始化：

1. /**
2.  * This is a pointer to the root of the DataTree. It is the source of truth,
3.  * but we usually use the nodes hashmap to find nodes in the tree.
4.  */
5. private DataNode root = new DataNode(null, new byte[0], -1L,new StatPersisted());
6.  
7. // create a /zookeeper filesystem that is the proc filesystem of zookeeper
8. private DataNode procDataNode = new DataNode(root, new byte[0], -1L,new StatPersisted());
9.  
10. // create a /zookeeper/quota node for maintaining quota properties for zookeeper
11. private DataNode quotaDataNode = new DataNode(procDataNode, new byte[0],-1L, new StatPersisted());
12.  
13. public DataTree() {
14.     // Rather than fight it, let root have an alias
15.     nodes.put("", root);
16.     nodes.put(rootZookeeper, root);
17.  
18.     // add the proc node and quota node
19.     root.addChild(procChildZookeeper);
20.     nodes.put(procZookeeper, procDataNode);
21.  
22.     procDataNode.addChild(quotaChildZookeeper);
23.     nodes.put(quotaZookeeper, quotaDataNode);
24. }

结构图为：

---

|---rootZookeeper = "/"

|---procZookeeper = "/zookeeper"

    |---procChildZookeeper ="zookeeper"

|---quotaZookeeper = "/zookeeper/quota"

    |---quotaChildZookeeper = "quota"

---

limitNode = "zookeeper_limits"

statNode = "zookeeper_stats"

---

|---DataNode root("/")

    |---root.children set<String>

        |---<Zookeeper>

|---DataNode procDataNode("/Zookeeper")

    |---procDataNode.children set<String>

        |---<quota>

|---DataNode procDataNode("/Zookeeper/quota")

    |---procDataNode.children set<String>

        |---<null>

---

|---nodes<String, DataNode>

    |---<"",root>

    |---<rootZookeeper,root>

    |---<procZookeeper, procDataNode>

    |---<quotaZookeeper, quotaDataNode>

五、节点操作

5.1 createNode过程

1. /**
2.  * @param path
3.  * @param data
4.  * @param acl
5.  * @param ephemeralOwner
6.  * the session id that owns this node. -1 indicates this is not
7.  * an ephemeral node.
8.  * @param zxid
9.  * @param time
10.  * @return the patch of the created node
11.  * @throws KeeperException
12.  */
13. public String createNode(String path, byte data[], List<ACL> acl,long ephemeralOwner, int parentCVersion, long zxid, long time)

详细代码：

  public String createNode(String path, byte data[], List<ACL> acl,
            long ephemeralOwner, int parentCVersion, long zxid, long time)
            throws KeeperException.NoNodeException,
            KeeperException.NodeExistsException {
        int lastSlash = path.lastIndexOf('/');
        String parentName = path.substring(0, lastSlash);
        String childName = path.substring(lastSlash + 1);
        StatPersisted stat = new StatPersisted();
        stat.setCtime(time);
        stat.setMtime(time);
        stat.setCzxid(zxid);
        stat.setMzxid(zxid);
        stat.setPzxid(zxid);
        stat.setVersion(0);
        stat.setAversion(0);
        stat.setEphemeralOwner(ephemeralOwner);
        DataNode parent = nodes.get(parentName);
        if (parent == null) {
            throw new KeeperException.NoNodeException();
        }
        synchronized (parent) {
            Set<String> children = parent.getChildren();
            if (children != null) {
                if (children.contains(childName)) {
                    throw new KeeperException.NodeExistsException();
                }
            }
            
            if (parentCVersion == -1) {
                parentCVersion = parent.stat.getCversion();
                parentCVersion++;
            }    
            parent.stat.setCversion(parentCVersion);
            parent.stat.setPzxid(zxid);
            Long longval = convertAcls(acl);
            DataNode child = new DataNode(parent, data, longval, stat);
            parent.addChild(childName);
            nodes.put(path, child);
            if (ephemeralOwner != 0) {
                HashSet<String> list = ephemerals.get(ephemeralOwner);
                if (list == null) {
                    list = new HashSet<String>();
                    ephemerals.put(ephemeralOwner, list);
                }
                synchronized (list) {
                    list.add(path);
                }
            }
        }
        // now check if its one of the zookeeper node child
        if (parentName.startsWith(quotaZookeeper)) {
            // now check if its the limit node
            if (Quotas.limitNode.equals(childName)) {
                // this is the limit node
                // get the parent and add it to the trie
                pTrie.addPath(parentName.substring(quotaZookeeper.length()));
            }
            if (Quotas.statNode.equals(childName)) {
                updateQuotaForPath(parentName
                        .substring(quotaZookeeper.length()));
            }
        }
        // also check to update the quotas for this node
        String lastPrefix;
        if((lastPrefix = getMaxPrefixWithQuota(path)) != null) {
            // ok we have some match and need to update
            updateCount(lastPrefix, 1);
            updateBytes(lastPrefix, data == null ? 0 : data.length);
        }
        dataWatches.triggerWatch(path, Event.EventType.NodeCreated);
        childWatches.triggerWatch(parentName.equals("") ? "/" : parentName,
                Event.EventType.NodeChildrenChanged);
        return path;
    }

具体创建过程如下：

　　① 创建StatPersisted stat元数据，并set各种成员变量；

　　② 创建DataNode child节点；

　　③ 解析父节点路径parentName，并通过DataNode parent = nodes.get(parentName)获取父节点，然后更新parent的pzxid、cversion、ephemeralOwner；

　　④ 将child放入parent的children列表中，以及放入DataTree的nodes中：parent.addChild(childName); nodes.put(path, child);

　　⑤ 如果是临时节点，需要保存到DataTree的ephemerals中，key是所属owner的sessionid；

　　⑥ 判断该节点是否/zookeeper/quota/zookeeper_limits或/zookeeper/quota/zookeeper_stat，如果是则？？？？；

　　⑦ 更新该节点的quota信息，即***/ zookeeper_stat节点内容；

　　⑧ 调用dataWatches.triggerWatch()触发该路径的Event.EventType.NodeCreated相关事件；

　　⑨ 调用childWatches.triggerWatch()触发父节点路径的Event.EventType.NodeChildrenChanged相关事件。

5.2 deleteNode过程

1. /**
2.  * remove the path from the datatree
3.  *
4.  * @param path
5.  * the path to of the node to be deleted
6.  * @param zxid
7.  * the current zxid
8.  * @throws KeeperException.NoNodeException
9.  */
10. public void deleteNode(String path, long zxid) throws KeeperException.NoNodeException {

 详细代码：

  public void deleteNode(String path, long zxid)
            throws KeeperException.NoNodeException {
        int lastSlash = path.lastIndexOf('/');
        String parentName = path.substring(0, lastSlash);
        String childName = path.substring(lastSlash + 1);
        DataNode node = nodes.get(path);
        if (node == null) {
            throw new KeeperException.NoNodeException();
        }
        nodes.remove(path);
        DataNode parent = nodes.get(parentName);
        if (parent == null) {
            throw new KeeperException.NoNodeException();
        }
        synchronized (parent) {
            parent.removeChild(childName);
            parent.stat.setPzxid(zxid);
            long eowner = node.stat.getEphemeralOwner();
            if (eowner != 0) {
                HashSet<String> nodes = ephemerals.get(eowner);
                if (nodes != null) {
                    synchronized (nodes) {
                        nodes.remove(path);
                    }
                }
            }
            node.parent = null;
        }
        if (parentName.startsWith(procZookeeper)) {
            // delete the node in the trie.
            if (Quotas.limitNode.equals(childName)) {
                // we need to update the trie
                // as well
                pTrie.deletePath(parentName.substring(quotaZookeeper.length()));
            }
        }

        // also check to update the quotas for this node
        String lastPrefix;
        if((lastPrefix = getMaxPrefixWithQuota(path)) != null) {
            // ok we have some match and need to update
            updateCount(lastPrefix, -1);
            int bytes = 0;
            synchronized (node) {
                bytes = (node.data == null ? 0 : -(node.data.length));
            }
            updateBytes(lastPrefix, bytes);
        }
        if (LOG.isTraceEnabled()) {
            ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,
                    "dataWatches.triggerWatch " + path);
            ZooTrace.logTraceMessage(LOG, ZooTrace.EVENT_DELIVERY_TRACE_MASK,
                    "childWatches.triggerWatch " + parentName);
        }
        Set<Watcher> processed = dataWatches.triggerWatch(path,
                EventType.NodeDeleted);
        childWatches.triggerWatch(path, EventType.NodeDeleted, processed);
        childWatches.triggerWatch(parentName.equals("") ? "/" : parentName,
                EventType.NodeChildrenChanged);
    }

具体的deleteNode过程如下：

　　① 根据DataNode node = nodes.get(path)获取该节点的DataNode；

　　② 根据DataNode parent = nodes.get(parentName)获取该节点的父节点；

　　③ 更新parent的children列表、cversion、pzxid、ephemeralOwner，如果是临时节点，还要更新DataTree的ephemerals；

　　④ 判断该节点是否/zookeeper/quota/zookeeper_limits或/zookeeper/quota/zookeeper_stat，如果是则？？？？；

　　⑤ 更新该节点的quota信息，即***/ zookeeper_stat节点内容；

　　⑥ 调用dataWatches.triggerWatch()触发该路径的Event.EventType.NodeDeleted相关事件；

　　⑦ 调用childWatches.triggerWatch()触发父节点路径的Event.EventType.NodeChildrenChanged相关事件

5.3 setData过程

1. public Stat setData(String path, byte data[], int version, long zxid,long time) throws KeeperException.NoNodeException {

详细代码：

  public Stat setData(String path, byte data[], int version, long zxid,
            long time) throws KeeperException.NoNodeException {
        Stat s = new Stat();
        DataNode n = nodes.get(path);
        if (n == null) {
            throw new KeeperException.NoNodeException();
        }
        byte lastdata[] = null;
        synchronized (n) {
            lastdata = n.data;
            n.data = data;
            n.stat.setMtime(time);
            n.stat.setMzxid(zxid);
            n.stat.setVersion(version);
            n.copyStat(s);
        }
        // now update if the path is in a quota subtree.
        String lastPrefix;
        if((lastPrefix = getMaxPrefixWithQuota(path)) != null) {
          this.updateBytes(lastPrefix, (data == null ? 0 : data.length)
              - (lastdata == null ? 0 : lastdata.length));
        }
        dataWatches.triggerWatch(path, EventType.NodeDataChanged);
        return s;
    }

具体的setData过程：

　　① 根据DataNode n = nodes.get(path)获取该节点DataNode；

　　② 更新n的data、mtime、mzxid、version信息；

　　③ 调用DataTree的updateBytes更新Quota信息；

　　④ 调用dataWatches.triggerWatch()触发该节点路径的Event.EventType. NodeDataChanged相关事件。

5.4 getData过程

1. public byte[] getData(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException

详细代码：

    public byte[] getData(String path, Stat stat, Watcher watcher)
            throws KeeperException.NoNodeException {
        DataNode n = nodes.get(path);
        if (n == null) {
            throw new KeeperException.NoNodeException();
        }
        synchronized (n) {
            n.copyStat(stat);
            if (watcher != null) {
                dataWatches.addWatch(path, watcher);
            }
            return n.data;
        }
    }

具体的getData过程如下：

　　① 根据DataNode n = nodes.get(path)获取该节点DataNode；

　　② 如果watcher参数不为NULL，调用dataWatches.addWatch()添加watcher；

　　③ 返回n的data信息。

5.5 statNode过程

1. public Stat statNode(String path, Watcher watcher) throws KeeperException.NoNodeException

5.6 getChildren 过程

1. public List<String> getChildren(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException

5.7 getCounts过程

1. /**
2.  * this method gets the count of nodes and the bytes under a subtree
3.  *
4.  * @param path
5.  * the path to be used
6.  * @param counts
7.  * the int count
8.  */
9. private void getCounts(String path, Counts counts)