ZooKeeper源码分析（一）---ZooKeeper接口介绍
一、Server角色每个Server在工作过程中有三种状态：
① LOOKING：当前Server不知道leader是谁，正在搜寻。
② LEADING：当前Server即为选举出来的leader。
③ FOLLOWING：leader已经选举出来，当前Server与之同步。
QuorumPeer定义了server的类型，其中ServerState表示server类型，LeanerType表示当ServerState为FOLLOWING时是参与者还是观察者，前者称为follower，后者称为observer。
代码如下：
[*]
public class QuorumPeer extends Thread implements QuorumStats.Provider{
[*]
public enum ServerState {
[*]
LOOKING, FOLLOWING, LEADING, OBSERVING;
[*]
}
[*]
public enum LearnerType {
[*]
PARTICIPANT, OBSERVER;
[*]
}
[*]
}
二、Znode类型
CreateMode中定义了四种节点类型，分别对应：
PERSISTENT：永久节点
EPHEMERAL：临时节点
ERSISTENT_SEQUENTIAL：永久节点、序列化
EPHEMERAL_SEQUENTIAL：临时节点、序列化
代码如下：
[*]
public enum CreateMode {
[*]
PERSISTENT (0, false, false),
[*]
PERSISTENT_SEQUENTIAL (2, false, true),
[*]
EPHEMERAL (1, true, false),
[*]
EPHEMERAL_SEQUENTIAL (3, true, true);
[*]
}
三、Stat类
Stat类定义znode节点的元信息，主要成员变量如下：
[*]
public class Stat implements Record {
[*]
private long czxid; // 创建时的zxid
[*]
private long mzxid; // 最新修改的zxid
[*]
private long ctime; // 创建时间 private
[*]
long mtime; // 修改时间
[*]
private int version; // 版本号，对应znode data
[*]
private int cversion; // 版本号，对应子znode
[*]
private int aversion; // 版本号，对应acl
[*]
private long ephemeralOwner; // 临时节点对应的client session id，默认为0
[*]
private int dataLength; // znode data长度
[*]
private int numChildren; // 子znode个数
[*]
private long pzxid; // 最新修改的zxid，貌似与mzxid重合了
[*]
}
注意：
Stat、StatPersisited、StatPersisitedV1三个类，其成员变量和逻辑基本一致，但StatPersisited类少了dataLength和numChildren属性，StatPersisitedV1类少了dataLength、 numChildren和pzxid属性，具体不同类用在什么地方待进一步分析。
三、DataNOde类
DataNode类记录了znode节点的所有信息，包括其父节点、子节点、数据内容、ACL信息、stat元数据等，主要成员变量如下：
[*]
public class DataNode implements Record {
[*]
DataNode parent;
[*]
byte data[];
[*]
Long acl;
[*]
public StatPersisted stat;
[*]
private Set<String> children = null;
[*]
}
需要注意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类：
[*]
/**
[*]
* This hashtable provides a fast lookup to the datanodes. The tree is the
[*]
* source of truth and is where all the locking occurs
[*]
*/
[*]
private final ConcurrentHashMap<String, DataNode> nodes =new ConcurrentHashMap<String, DataNode>();
[*]
private final WatchManager dataWatches = new WatchManager();
[*]
private final WatchManager childWatches = new WatchManager();
[*]
[*]
private static final String rootZookeeper = "/"; //ZooKeeper树的根节点
[*]
[*]
private static final String procZookeeper = Quotas.procZookeeper;// ZooKeeper节点，作为管理和状态节点
[*]
private static final String procChildZookeeper = procZookeeper.substring(1);//存储根节点的子节点的字符串
[*]
//the zookeeper quota node that acts as the quota management node for zookeeper
[*]
private static final String quotaZookeeper = Quotas.quotaZookeeper; //ZooKeeper quota节点，作为ZooKeeper的配额管理节点
[*]
private static final String quotaChildZookeeper = quotaZookeeper.substring(procZookeeper.length() + 1); // 存储ZooKeeper节点的子节点字符串
[*]
[*]
private final PathTrie pTrie = new PathTrie(); //path trie跟踪在DataTree中的quota节点
[*]
[*]
//该hashtable列出了一个会话的临时节点路径
[*]
private final Map<Long, HashSet<String>> ephemerals =new ConcurrentHashMap<Long, HashSet<String>>();
[*]
//this is map from longs to acl's. It saves acl's being stored for each datanode.
[*]
public final Map<Long, List<ACL>> longKeyMap =new HashMap<Long, List<ACL>>();
[*]
//this a map from acls to long.
[*]
public final Map<List<ACL>, Long> aclKeyMap =new HashMap<List<ACL>, Long>();
[*]
[*]
//在DataTree中acls的数量
[*]
protected long aclIndex = 0;
（2）Quota类：
[*]
public class Quotas {
[*]
[*]
// the zookeeper nodes that acts as the management and status node
[*]
public static final String procZookeeper = "/zookeeper";
[*]
[*]
// the zookeeper quota node that acts as the quota management node for zookeeper
[*]
public static final String quotaZookeeper = "/zookeeper/quota";
[*]
[*]
//the limit node that has the limit of a subtree
[*]
public static final String limitNode = "zookeeper_limits";
[*]
[*]
//the stat node that monitors the limit of a subtree.
[*]
public static final String statNode = "zookeeper_stats";
[*]
[*]
/**
[*]
* return the quota path associated with this
[*]
* prefix
[*]
* @param path the actual path in zookeeper.
[*]
* @return the limit quota path
[*]
*/
[*]
public static String quotaPath(String path) {
[*]
return quotaZookeeper + path +
[*]
"/" + limitNode;
[*]
}
[*]
[*]
/**
[*]
* return the stat quota path associated with this
[*]
* prefix.
[*]
* @param path the actual path in zookeeper
[*]
* @return the stat quota path
[*]
*/
[*]
public static String statPath(String path) {
[*]
return quotaZookeeper + path + "/" +
[*]
statNode;
[*]
}
[*]
}
（2）StatsTrack类
[*]
//a class that represents the stats associated with quotas
[*]
public class StatsTrack {
[*]
private int count;
[*]
private long bytes;
[*]
private String countStr = "count";
[*]
private String byteStr = "bytes";
[*]
[*]
public StatsTrack() {
[*]
this(null);
[*]
}
[*]
/**
[*]
* the stat string should be of the form count=int,bytes=long
[*]
* if stats is called with null the count and bytes are initialized
[*]
* to -1.
[*]
* @param stats the stat string to be intialized with
[*]
*/
[*]
public StatsTrack(String stats) {
[*]
if (stats == null) {
[*]
stats = "count=-1,bytes=-1";
[*]
}
[*]
String[] split = stats.split(",");
[*]
if (split.length != 2) {
[*]
throw new IllegalArgumentException("invalid string " + stats);
[*]
}
[*]
count = Integer.parseInt(split.split("="));
[*]
bytes = Long.parseLong(split.split("="));
[*]
}
[*]
[*]
[*]
/**
[*]
* get the count of nodes allowed as part of quota
[*]
*
[*]
* @return the count as part of this string
[*]
*/
[*]
public int getCount() {
[*]
return this.count;
[*]
}
[*]
[*]
/**
[*]
* set the count for this stat tracker.
[*]
*
[*]
* @param count
[*]
* the count to set with
[*]
*/
[*]
public void setCount(int count) {
[*]
this.count = count;
[*]
}
[*]
[*]
/**
[*]
* get the count of bytes allowed as part of quota
[*]
*
[*]
* @return the bytes as part of this string
[*]
*/
[*]
public long getBytes() {
[*]
return this.bytes;
[*]
}
[*]
[*]
/**
[*]
* set teh bytes for this stat tracker.
[*]
*
[*]
* @param bytes
[*]
* the bytes to set with
[*]
*/
[*]
public void setBytes(long bytes) {
[*]
this.bytes = bytes;
[*]
}
[*]
[*]
@Override
[*]
/*
[*]
* returns the string that maps to this stat tracking.
[*]
*/
[*]
public String toString() {
[*]
return countStr + "=" + count + "," + byteStr + "=" + bytes;
[*]
}
[*]
}
四、DataTree初始化
DataTree初始化要完成的工作，需要建立系统节点，包括/、/zookeeper、/zookeeper/quota三个znode。
下面先看一下DataNode的构造函数如下：
[*]
public DataNode(DataNode parent, byte data[], Long acl, StatPersisted stat) {
[*]
this.parent = parent;
[*]
this.data = data;
[*]
this.acl = acl;
[*]
this.stat = stat;
[*]
}
Datatree初始化：
[*]
/**
[*]
* This is a pointer to the root of the DataTree. It is the source of truth,
[*]
* but we usually use the nodes hashmap to find nodes in the tree.
[*]
*/
[*]
private DataNode root = new DataNode(null, new byte, -1L,new StatPersisted());
[*]
[*]
// create a /zookeeper filesystem that is the proc filesystem of zookeeper
[*]
private DataNode procDataNode = new DataNode(root, new byte, -1L,new StatPersisted());
[*]
[*]
// create a /zookeeper/quota node for maintaining quota properties for zookeeper
[*]
private DataNode quotaDataNode = new DataNode(procDataNode, new byte,-1L, new StatPersisted());
[*]
[*]
public DataTree() {
[*]
// Rather than fight it, let root have an alias
[*]
nodes.put("", root);
[*]
nodes.put(rootZookeeper, root);
[*]
[*]
// add the proc node and quota node
[*]
root.addChild(procChildZookeeper);
[*]
nodes.put(procZookeeper, procDataNode);
[*]
[*]
procDataNode.addChild(quotaChildZookeeper);
[*]
nodes.put(quotaZookeeper, quotaDataNode);
[*]
}
结构图为：
|---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过程
[*]
/**
[*]
* @param path
[*]
* @param data
[*]
* @param acl
[*]
* @param ephemeralOwner
[*]
* the session id that owns this node. -1 indicates this is not
[*]
* an ephemeral node.
[*]
* @param zxid
[*]
* @param time
[*]
* @return the patch of the created node
[*]
* @throws KeeperException
[*]
*/
[*]
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;
}
View Code 具体创建过程如下：
① 创建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过程
[*]
/**
[*]
* remove the path from the datatree
[*]
*
[*]
* @param path
[*]
* the path to of the node to be deleted
[*]
* @param zxid
[*]
* the current zxid
[*]
* @throws KeeperException.NoNodeException
[*]
*/
[*]
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);
}
View Code 具体的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过程
[*]
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;
}
View Code 具体的setData过程：
① 根据DataNode n = nodes.get(path)获取该节点DataNode；
② 更新n的data、mtime、mzxid、version信息；
③ 调用DataTree的updateBytes更新Quota信息；
④ 调用dataWatches.triggerWatch()触发该节点路径的Event.EventType. NodeDataChanged相关事件。
5.4 getData过程
[*]
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;
}
}
View Code 具体的getData过程如下：
① 根据DataNode n = nodes.get(path)获取该节点DataNode；
② 如果watcher参数不为NULL，调用dataWatches.addWatch()添加watcher；
③ 返回n的data信息。
5.5 statNode过程
[*]
public Stat statNode(String path, Watcher watcher) throws KeeperException.NoNodeException
5.6 getChildren 过程
[*]
public List<String> getChildren(String path, Stat stat, Watcher watcher) throws KeeperException.NoNodeException
5.7 getCounts过程
[*]
/**
[*]
* this method gets the count of nodes and the bytes under a subtree
[*]
*
[*]
* @param path
[*]
* the path to be used
[*]
* @param counts
[*]
* the int count
[*]
*/
[*]
private void getCounts(String path, Counts counts)
页:
[1]