本文将从leader处理器入手，详细分析node的增删改查流程及监听器原理。
回顾数据读写流程
leader
- ZookeeperServer.processPacket封装Request并提交给业务处理器
- LeaderRequestProcessor做本地事务升级
- PrepRequestProcessor做事务准备
- ProposalRequestProcessor事务操作发proposal给follower节点，持久化到log文件
- CommitProcessor读请求直接转发给下游处理器，事务操作等待到了quorum状态转发给下游处理器
- ToBeAppliedRequestProcessor清理toBeApplied集
- FinalRequestProcessor将事务写到ZKDatabase中，给客户端发响应
follower
- 处理PROPOSAL：使用SyncRequestProcessor处理器持久化，之后SendAckRequestProcessor给leader发ack
- 处理COMMIT：提交给CommitProcessor处理器，之后FinalRequestProcessor将事务写到ZKDatabase中
创建node
涉及create、create2、createContainer、createTTL等命令。
PrepRequestProcessor事务准备
反序列化请求参数
switch (request.type) {
case OpCode.createContainer:
case OpCode.create:
case OpCode.create2:
CreateRequest create2Request = request.readRequestRecord(CreateRequest::new);
pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request);
break;
case OpCode.createTTL:
// 默认不支持ttl
CreateTTLRequest createTtlRequest = request.readRequestRecord(CreateTTLRequest::new);
pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest);
break;
// ...
CreateRequest封装创建node的参数：
public class CreateRequest implements Record {
private String path;
private byte[] data;
private java.util.List<org.apache.zookeeper.data.ACL> acl;
private int flags;
}
CreateTTLRequest封装创建node加ttl的参数：
public class CreateTTLRequest implements Record {
private String path;
private byte[] data;
private java.util.List<org.apache.zookeeper.data.ACL> acl;
private int flags;
private long ttl;
}
事务准备
protected void pRequest2Txn(int type, long zxid, Request request, Record record)
throws KeeperException, IOException, RequestProcessorException {
// ...
switch (type) {
case OpCode.create:
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer: {
pRequest2TxnCreate(type, request, record);
break;
}
// ...
}
}
private void pRequest2TxnCreate(
int type, Request request, Record record) throws IOException, KeeperException {
int flags;
String path;
List<ACL> acl;
byte[] data;
long ttl;
if (type == OpCode.createTTL) {
CreateTTLRequest createTtlRequest = (CreateTTLRequest) record;
// 给flags等参数赋值
} else {
CreateRequest createRequest = (CreateRequest) record;
// 给flags等参数赋值
ttl = -1;
}
// CreateMode:
// PERSISTENT, PERSISTENT_SEQUENTIAL, EPHEMERAL, EPHEMERAL_SEQUENTIAL,
// CONTAINER, PERSISTENT_WITH_TTL, PERSISTENT_SEQUENTIAL_WITH_TTL
CreateMode createMode = CreateMode.fromFlag(flags);
// 验证临时节点、ttl参数、检查session
// 默认不支持ttl
validateCreateRequest(path, createMode, request, ttl);
String parentPath = validatePathForCreate(path, request.sessionId); // 父节点path
List<ACL> listACL = fixupACL(path, request.authInfo, acl); // 请求携带的权限
ChangeRecord parentRecord = getRecordForPath(parentPath); // 得到父节点
// 验证CREATE权限
zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo, path, listACL);
int parentCVersion = parentRecord.stat.getCversion();
if (createMode.isSequential()) { // 顺序节点
// 例如/users/admin0000000001
path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
}
validatePath(path, request.sessionId);
// 略
boolean ephemeralParent =
EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
// 父节点不可以是临时节点
int newCversion = parentRecord.stat.getCversion() + 1; // 父节点的childVersion++
// 检查字节限额
zks.checkQuota(path, null, data, OpCode.create);
// 不同类型创建对应的Txn对象
if (type == OpCode.createContainer) {
request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
} else if (type == OpCode.createTTL) {
request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
} else {
request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(), newCversion));
}
TxnHeader hdr = request.getHdr();
long ephemeralOwner = 0;
if (createMode.isContainer()) {
ephemeralOwner = EphemeralType.CONTAINER_EPHEMERAL_OWNER;
} else if (createMode.isTTL()) {
ephemeralOwner = EphemeralType.TTL.toEphemeralOwner(ttl);
} else if (createMode.isEphemeral()) {
ephemeralOwner = request.sessionId; // 临时节点使用sessionId
}
// czxid(created),mzxid(modified),ctime,mtime,version,cversion(childVersion),
// aversion(aclVersion),ephemeralOwner,pzxid(lastModifiedChildren)
StatPersisted s = DataTree.createStat(hdr.getZxid(), hdr.getTime(), ephemeralOwner);
// 父节点
parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
parentRecord.childCount++;
parentRecord.stat.setCversion(newCversion);
parentRecord.stat.setPzxid(request.getHdr().getZxid());
parentRecord.precalculatedDigest = precalculateDigest(
DigestOpCode.UPDATE, parentPath, parentRecord.data, parentRecord.stat);
addChangeRecord(parentRecord);
// 新增节点
ChangeRecord nodeRecord = new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL);
nodeRecord.data = data;
nodeRecord.precalculatedDigest = precalculateDigest(DigestOpCode.ADD, path, nodeRecord.data, s);
setTxnDigest(request, nodeRecord.precalculatedDigest);
addChangeRecord(nodeRecord);
}
protected void addChangeRecord(ChangeRecord c) {
synchronized (zks.outstandingChanges) {
zks.outstandingChanges.add(c);
zks.outstandingChangesForPath.put(c.path, c);
}
}
outstandingChanges保存未提交的事务变化，比如在生成顺序节点时需要使用cversion值，但是在事务提交到ZKDatabase之前，库里面的值是旧的，所以在上面的代码中，是从outstandingChanges查找节点，给cversion++后再生成顺序节点。
在事务提交之后，才会清理outstandingChanges集。
ProposalRequestProcessor发Proposal
public void processRequest(Request request) throws RequestProcessorException {
if (request instanceof LearnerSyncRequest) { // sync命令流程，暂不分析
zks.getLeader().processSync((LearnerSyncRequest) request);
} else {
if (shouldForwardToNextProcessor(request)) {
nextProcessor.processRequest(request); // 提交给下游处理器
}
if (request.getHdr() != null) { // 事务操作需要发proposal并写磁盘
try {
zks.getLeader().propose(request);
} catch (XidRolloverException e) {
throw new RequestProcessorException(e.getMessage(), e);
}
// 把事务log写到文件中
// 之后通过AckRequestProcessor处理器给leader ack
syncProcessor.processRequest(request);
}
}
}
CommitProcessor提交事务
public void processRequest(Request request) {
request.commitProcQueueStartTime = Time.currentElapsedTime();
queuedRequests.add(request); // 所有请求队列
if (needCommit(request)) { // 需要提交的请求进入到写队列
queuedWriteRequests.add(request);
numWriteQueuedRequests.incrementAndGet();
} else {
numReadQueuedRequests.incrementAndGet();
}
wakeup();
}
run方法对比queuedRequests、queuedWriteRequests、committedRequests这几个队列，将提交成功的请求或读请求转发给下游的ToBeAppliedRequestProcessor处理器。
FinalRequestProcessor应用事务
该处理器位于处理器链的末尾，负责将事务应用到ZKDatabase、查询数据、返回响应。
applyRequest
该方法将事务应用到ZKDatabase中：
private ProcessTxnResult applyRequest(Request request) {
// 应用事务
ProcessTxnResult rc = zks.processTxn(request);
// closeSession
// metrics
return rc;
}
zks.processTxn负责处理session、处理事务、清理outstandingChanges集。重点看一下处理事务的步骤。
processTxn
public ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn) {
ProcessTxnResult rc = new ProcessTxnResult();
try {
rc.clientId = header.getClientId();
rc.cxid = header.getCxid();
rc.zxid = header.getZxid();
rc.type = header.getType();
rc.err = 0;
rc.multiResult = null;
switch (header.getType()) {
case OpCode.create:
CreateTxn createTxn = (CreateTxn) txn;
rc.path = createTxn.getPath();
createNode(
createTxn.getPath(),
createTxn.getData(),
createTxn.getAcl(),
createTxn.getEphemeral() ? header.getClientId() : 0,
createTxn.getParentCVersion(),
header.getZxid(),
header.getTime(),
null);
break;
case OpCode.create2:
CreateTxn create2Txn = (CreateTxn) txn;
rc.path = create2Txn.getPath();
Stat stat = new Stat();
createNode(
create2Txn.getPath(),
create2Txn.getData(),
create2Txn.getAcl(),
create2Txn.getEphemeral() ? header.getClientId() : 0,
create2Txn.getParentCVersion(),
header.getZxid(),
header.getTime(),
stat);
rc.stat = stat;
break;
case OpCode.createTTL:
CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
rc.path = createTtlTxn.getPath();
stat = new Stat();
createNode(
createTtlTxn.getPath(),
createTtlTxn.getData(),
createTtlTxn.getAcl(),
EphemeralType.TTL.toEphemeralOwner(createTtlTxn.getTtl()),
createTtlTxn.getParentCVersion(),
header.getZxid(),
header.getTime(),
stat);
rc.stat = stat;
break;
case OpCode.createContainer:
CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
rc.path = createContainerTxn.getPath();
stat = new Stat();
createNode(
createContainerTxn.getPath(),
createContainerTxn.getData(),
createContainerTxn.getAcl(),
EphemeralType.CONTAINER_EPHEMERAL_OWNER,
createContainerTxn.getParentCVersion(),
header.getZxid(),
header.getTime(),
stat);
rc.stat = stat;
break;
// ...
}
}
// ...
}
createNode
public void createNode(final String path, byte[] data, List<ACL> acl, long ephemeralOwner, int parentCVersion, long zxid, long time, Stat outputStat) throws NoNodeException, NodeExistsException {
int lastSlash = path.lastIndexOf('/');
String parentName = path.substring(0, lastSlash);
String childName = path.substring(lastSlash + 1);
StatPersisted stat = createStat(zxid, time, ephemeralOwner);
DataNode parent = nodes.get(parentName); // 父节点必须存在
if (parent == null) {
throw new NoNodeException();
}
synchronized (parent) {
Long acls = aclCache.convertAcls(acl);
Set<String> children = parent.getChildren();
if (children.contains(childName)) { // 节点不能存在
throw new NodeExistsException();
}
nodes.preChange(parentName, parent);
if (parentCVersion == -1) { // childVersion++
parentCVersion = parent.stat.getCversion();
parentCVersion++;
}
if (parentCVersion > parent.stat.getCversion()) {
parent.stat.setCversion(parentCVersion);
parent.stat.setPzxid(zxid);
}
// 创建node
DataNode child = new DataNode(data, acls, stat);
parent.addChild(childName);
nodes.postChange(parentName, parent);
nodeDataSize.addAndGet(getNodeSize(path, child.data));
nodes.put(path, child); // 维护NodeHashMap
// 处理临时节点
EphemeralType ephemeralType = EphemeralType.get(ephemeralOwner);
if (ephemeralType == EphemeralType.CONTAINER) {
containers.add(path);
} else if (ephemeralType == EphemeralType.TTL) {
ttls.add(path);
} else if (ephemeralOwner != 0) {
HashSet<String> list = ephemerals.computeIfAbsent(ephemeralOwner, k -> new HashSet<>());
synchronized (list) {
list.add(path);
}
}
// 返回节点stat
if (outputStat != null) {
child.copyStat(outputStat);
}
}
// now check if its one of the zookeeper node child 略
// 触发NodeCreated监听
dataWatches.triggerWatch(path, Event.EventType.NodeCreated, zxid);
// 触发父节点的NodeChildrenChanged监听
childWatches.triggerWatch(
parentName.equals("") ? "/" : parentName, Event.EventType.NodeChildrenChanged, zxid);
}
返回响应
case OpCode.create: {
lastOp = "CREA";
rsp = new CreateResponse(rc.path); // 创建Response
err = Code.get(rc.err); // processTxn的err
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
case OpCode.create2:
case OpCode.createTTL:
case OpCode.createContainer: {
lastOp = "CREA";
rsp = new Create2Response(rc.path, rc.stat); // 创建Response
err = Code.get(rc.err); // processTxn的err
requestPathMetricsCollector.registerRequest(request.type, rc.path);
break;
}
最后会使用cnxn把响应返回给客户端：
ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
cnxn.sendResponse(hdr, rsp, "response");
EphemeralType
- VOID
- NORMAL
- CONTAINER
- TTL
ephemeralOwner标识znode是临时的，以及哪个会话创建了该节点。通过zookeeper.extendedTypesEnabled属性可以启用ttl节点等扩展功能。ephemeralOwner的"特殊位"用于表示启用了哪个功能，而ephemeral Owner的剩余位是特定于功能的。
当zookeeper.extendedTypesEnabled为true时，将启用扩展类型。扩展ephemeralOwner填充高8位(0xff00000000000000L)，高8位之后的两个字节用于表示ephemeralOwner扩展特征，剩余5个字节由该功能指定，可用于所需的任何目的。
目前，唯一扩展功能是TTL节点，扩展特征值为0。对于TTL节点，ephemeralOwner具有0xff的高8位，接下来2个字节是0，后面的5个字节是以毫秒为单位的ttl值。因此ttl值为1毫秒的ephemeralOwner是0xff00000000000001。
要添加新的扩展功能：
- 向枚举添加新名称
- 在ttl之后，定义常量extended_BIT_xxxx，即0x0001
- 通过静态初始值设定项向extendedFeatureMap添加映射
注意：从技术上讲，容器节点也是扩展类型，但由于它是在该功能之前实现的，因此被特别表示。根据定义，只有高位集(0x8000000000000000L)的临时所有者是容器节点(无论是否启用扩展类型)。
ttl节点
- 默认不开启，使用
- Added in 3.5.3
- 创建PERSISTENT或PERSISTENT_SEQUENTIAL节点时，可以设置以毫秒为单位的ttl。如果znode没有在ttl内修改，并且没有子节点，它将在将来的某个时候成为服务器删除的候选节点
container节点
- Added in 3.5.3
- 当container节点的最后一个子节点被删除时，该container节点将成为服务器在未来某个时候删除的候选节点
Stat类
封装节点属性，字段如下：
- czxid
The zxid of the change that caused this znode to be created.
- mzxid
The zxid of the change that last modified this znode.
- pzxid
The zxid of the change that last modified children of this znode.
- ctime
The time in milliseconds from epoch when this znode was created.
- mtime
The time in milliseconds from epoch when this znode was last modified.
- version
The number of changes to the data of this znode.
- cversion
The number of changes to the children of this znode.
- aversion
The number of changes to the ACL of this znode.
- ephemeralOwner
The session id of the owner of this znode if the znode is an ephemeral node. If it is not an ephemeral node, it will be zero.
- dataLength
The length of the data field of this znode.
- numChildren
The number of children of this znode.
删除node
涉及delete、deleteContainer等命令。
PrepRequestProcessor事务准备
反序列化请求参数
private void pRequestHelper(Request request) {
try {
switch (request.type) {
// ...
case OpCode.deleteContainer:
DeleteContainerRequest deleteContainerRequest =
request.readRequestRecord(DeleteContainerRequest::new);
pRequest2Txn(request.type, zks.getNextZxid(), request, deleteContainerRequest);
break;
case OpCode.delete:
DeleteRequest deleteRequest = request.readRequestRecord(DeleteRequest::new);
pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest);
break;
}
// ...
}
}
DeleteContainerRequest类：
public class DeleteContainerRequest implements Record {
private String path;
}
DeleteRequest类：
public class DeleteRequest implements Record {
private String path;
private int version;
}
事务准备
protected void pRequest2Txn(int type, long zxid, Request request,
Record record) throws KeeperException, IOException, RequestProcessorException {
// 略
switch (type) {
// 略
case OpCode.deleteContainer: {
DeleteContainerRequest txn = (DeleteContainerRequest) record;
String path = txn.getPath();
String parentPath = getParentPathAndValidate(path);
ChangeRecord nodeRecord = getRecordForPath(path); // 获取待删除节点
if (nodeRecord.childCount > 0) { // 有子节点不允许删除
throw new KeeperException.NotEmptyException(path);
}
if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL) {
throw new KeeperException.BadVersionException(path);
}
ChangeRecord parentRecord = getRecordForPath(parentPath); // 获取父节点
request.setTxn(new DeleteTxn(path));
// addChangeRecord 略
break;
}
case OpCode.delete:
zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
DeleteRequest deleteRequest = (DeleteRequest) record;
String path = deleteRequest.getPath();
String parentPath = getParentPathAndValidate(path);
ChangeRecord parentRecord = getRecordForPath(parentPath); // 获取父节点
// 检查DELETE权限
zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo, path, null);
ChangeRecord nodeRecord = getRecordForPath(path); // 获取待删除节点
checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path); // 检查version
if (nodeRecord.childCount > 0) { // 有子节点不允许删除
throw new KeeperException.NotEmptyException(path);
}
request.setTxn(new DeleteTxn(path));
// addChangeRecord 略
break;
}
}
FinalRequestProcessor应用事务
processTxn
public ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn) {
ProcessTxnResult rc = new ProcessTxnResult();
try {
rc.clientId = header.getClientId();
rc.cxid = header.getCxid();
rc.zxid = header.getZxid();
rc.type = header.getType();
rc.err = 0;
rc.multiResult = null;
switch (header.getType()) {
// ...
case OpCode.delete:
case OpCode.deleteContainer:
DeleteTxn deleteTxn = (DeleteTxn) txn;
rc.path = deleteTxn.getPath();
deleteNode(deleteTxn.getPath(), header.getZxid());
break;
}
// ...
}
}
deleteNode
public void deleteNode(String path, long zxid) throws NoNodeException {
int lastSlash = path.lastIndexOf('/');
String parentName = path.substring(0, lastSlash);
String childName = path.substring(lastSlash + 1);
DataNode parent = nodes.get(parentName);
if (parent == null) { // 获取父节点且必须存在
throw new NoNodeException();
}
synchronized (parent) {
nodes.preChange(parentName, parent);
parent.removeChild(childName);
if (zxid > parent.stat.getPzxid()) {
parent.stat.setPzxid(zxid); // The zxid of the change that last modified children of this znode
}
nodes.postChange(parentName, parent);
}
DataNode node = nodes.get(path); // 获取删除节点
if (node == null) {
throw new NoNodeException();
}
nodes.remove(path); // 从NodeHashMap删除
synchronized (node) { // 移除权限
aclCache.removeUsage(node.acl);
nodeDataSize.addAndGet(-getNodeSize(path, node.data));
}
// 移除临时节点、container、ttl等缓存
synchronized (parent) {
long owner = node.stat.getEphemeralOwner();
EphemeralType ephemeralType = EphemeralType.get(owner);
if (ephemeralType == EphemeralType.CONTAINER) {
containers.remove(path);
} else if (ephemeralType == EphemeralType.TTL) {
ttls.remove(path);
} else if (owner != 0) {
Set<String> nodes = ephemerals.get(owner);
if (nodes != null) {
synchronized (nodes) {
nodes.remove(path);
}
}
}
}
// 略
// 触发NodeDeleted监听
WatcherOrBitSet processed = dataWatches.triggerWatch(path, EventType.NodeDeleted, zxid);
childWatches.triggerWatch(path, EventType.NodeDeleted, zxid, processed);
// 触发父节点的NodeChildrenChanged监听
childWatches.triggerWatch(
"".equals(parentName) ? "/" : parentName, EventType.NodeChildrenChanged, zxid);
}
设置node数据
PrepRequestProcessor事务准备
反序列化请求参数
private void pRequestHelper(Request request) {
try {
switch (request.type) {
// ...
case OpCode.setData:
SetDataRequest setDataRequest = request.readRequestRecord(SetDataRequest::new);
pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest);
break;
// other case
}
}
// ...
}
SetDataRequest类：
public class SetDataRequest implements Record {
private String path;
private byte[] data;
private int version;
}
事务准备
protected void pRequest2Txn(int type, long zxid, Request request,
Record record) throws KeeperException, IOException, RequestProcessorException {
// 略
switch (type) {
// ...
case OpCode.setData:
zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
SetDataRequest setDataRequest = (SetDataRequest) record;
path = setDataRequest.getPath();
validatePath(path, request.sessionId);
nodeRecord = getRecordForPath(path); // 获取节点对象
// 检查权限
zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);
// 检查字节限额
zks.checkQuota(path, nodeRecord.data, setDataRequest.getData(), OpCode.setData);
// version++
int newVersion = checkAndIncVersion(
nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
// 创建SetDataTxn
request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
// addChangeRecord
break;
// other case
}
}
FinalRequestProcessor应用事务
processTxn
public ProcessTxnResult processTxn(TxnHeader header, Record txn, boolean isSubTxn) {
ProcessTxnResult rc = new ProcessTxnResult();
try {
rc.clientId = header.getClientId();
rc.cxid = header.getCxid();
rc.zxid = header.getZxid();
rc.type = header.getType();
rc.err = 0;
rc.multiResult = null;
switch (header.getType()) {
// other case
case OpCode.setData:
SetDataTxn setDataTxn = (SetDataTxn) txn;
rc.path = setDataTxn.getPath();
rc.stat = setData(
setDataTxn.getPath(),
setDataTxn.getData(),
setDataTxn.getVersion(),
header.getZxid(),
header.getTime());
break;
// other case
}
}
// ...
}
setData
public Stat setData(String path, byte[] data, int version,
long zxid, long time) throws NoNodeException {
Stat s = new Stat();
DataNode n = nodes.get(path);
if (n == null) { // 检查节点存在
throw new NoNodeException();
}
byte[] lastData;
synchronized (n) {
lastData = n.data;
nodes.preChange(path, n);
n.data = data; // 节点数据
n.stat.setMtime(time); // 修改时间
n.stat.setMzxid(zxid); // 修改zxid
n.stat.setVersion(version); // 版本
n.copyStat(s);
nodes.postChange(path, n);
}
// 略
// 触发NodeDataChanged监听
dataWatches.triggerWatch(path, EventType.NodeDataChanged, zxid);
return s;
}
查询node数据
PrepRequestProcessor验证session
经过该处理器时，只做session验证。
之后的ProposalRequestProcessor、CommitProcessor、ToBeAppliedRequestProcessor都是直接通过，不做事务处理，直接交给FinalRequestProcessor处理器查询数据、发送响应。
FinalRequestProcessor查询数据
使用handleGetDataRequest方法查询数据：
private Record handleGetDataRequest(
Record request, ServerCnxn cnxn, List<Id> authInfo) throws KeeperException, IOException {
GetDataRequest getDataRequest = (GetDataRequest) request;
String path = getDataRequest.getPath();
DataNode n = zks.getZKDatabase().getNode(path);
if (n == null) {
throw new KeeperException.NoNodeException();
}
// 检查权限
zks.checkACL(cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, authInfo, path, null);
Stat stat = new Stat();
// 查询数据
// 如果watcher参数不为null会给path添加一个监听器
byte[] b = zks.getZKDatabase().getData(path, stat, getDataRequest.getWatch() ? cnxn : null);
return new GetDataResponse(b, stat);
}
GetDataRequest类：
public class GetDataRequest implements Record {
private String path;
private boolean watch;
}
节点监听
addWatch命令
case OpCode.addWatch: {
lastOp = "ADDW";
AddWatchRequest addWatcherRequest = request.readRequestRecord(AddWatchRequest::new);
// 最终使用DataTree的addWatch方法注册监听器
// cnxn是ServerCnxn对象，实现了Watcher接口
zks.getZKDatabase().addWatch(addWatcherRequest.getPath(), cnxn, addWatcherRequest.getMode());
rsp = new ErrorResponse(0);
break;
}
DataTree的addWatch方法
public void addWatch(String basePath, Watcher watcher, int mode) {
// PERSISTENT|PERSISTENT_RECURSIVE
WatcherMode watcherMode = WatcherMode.fromZooDef(mode);
// dataWatches和childWatches是WatchManager类型对象
dataWatches.addWatch(basePath, watcher, watcherMode);
if (watcherMode != WatcherMode.PERSISTENT_RECURSIVE) {
childWatches.addWatch(basePath, watcher, watcherMode);
}
}
WatcherMode枚举
public enum WatcherMode {
STANDARD(false, false),
PERSISTENT(true, false), // persistent=0
PERSISTENT_RECURSIVE(true, true), // persistentRecursive=1
;
}
PERSISTENT和PERSISTENT_RECURSIVE是3.6.0版本新增的特性。
Watcher接口
实现类需要实现process方法：
void process(WatchedEvent event);
WatchedEvent代表一个监听事件：
public class WatchedEvent {
// 当前zk服务器的状态
private final KeeperState keeperState;
// NodeCreated|NodeDeleted|NodeDataChanged|NodeChildrenChanged等
private final EventType eventType;
private final String path;
private final long zxid;
}
重要的实现类：
- NIOServerCnxn
- NettyServerCnxn
WatchManager类
This class manages watches. It allows watches to be associated with a string and removes watchers and their watches in addition to managing triggers.
核心字段：
// path -> Watcher集
private final Map<String, Set<Watcher>> watchTable = new HashMap<>();
// Watcher -> path->WatchStats(PERSISTENT|STANDARD + PERSISTENT|PERSISTENT_RECURSIVE等)
private final Map<Watcher, Map<String, WatchStats>> watch2Paths = new HashMap<>();
WatchStats类
public final class WatchStats {
private static final WatchStats[] WATCH_STATS = new WatchStats[] {
new WatchStats(0), // NONE
new WatchStats(1), // STANDARD
new WatchStats(2), // PERSISTENT
new WatchStats(3), // STANDARD + PERSISTENT
new WatchStats(4), // PERSISTENT_RECURSIVE
new WatchStats(5), // STANDARD + PERSISTENT_RECURSIVE
new WatchStats(6), // PERSISTENT + PERSISTENT_RECURSIVE
new WatchStats(7), // STANDARD + PERSISTENT + PERSISTENT_RECURSIVE
};
public static final WatchStats NONE = WATCH_STATS[0];
private final int flags;
private WatchStats(int flags) {
this.flags = flags;
}
private static int modeToFlag(WatcherMode mode) {
// mode = STANDARD; return 1 << 0 = 1(0001)
// mode = PERSISTENT; return 1 << 1 = 2(0010)
// mode = PERSISTENT_RECURSIVE; return 1 << 2 = (0100)
return 1 << mode.ordinal();
}
public WatchStats addMode(WatcherMode mode) {
int flags = this.flags | modeToFlag(mode); // |计算保留多种状态
return WATCH_STATS[flags];
}
public WatchStats removeMode(WatcherMode mode) {
int mask = ~modeToFlag(mode); // 取反
int flags = this.flags & mask;
if (flags == 0) {
return NONE;
}
return WATCH_STATS[flags];
}
// ...
}
addWatch
public synchronized boolean addWatch(String path, Watcher watcher, WatcherMode watcherMode) {
Set<Watcher> list = watchTable.get(path);
if (list == null) {
list = new HashSet<>(4);
watchTable.put(path, list);
}
list.add(watcher); // 添加watchTable
Map<String, WatchStats> paths = watch2Paths.get(watcher);
if (paths == null) {
// cnxns typically have many watches, so use default cap here
paths = new HashMap<>();
watch2Paths.put(watcher, paths);
}
WatchStats stats = paths.getOrDefault(path, WatchStats.NONE);
WatchStats newStats = stats.addMode(watcherMode);
if (newStats != stats) {
paths.put(path, newStats);
if (watcherMode.isRecursive()) {
++recursiveWatchQty;
}
return true;
}
return false;
}
triggerWatch
public WatcherOrBitSet triggerWatch(String path, EventType type, long zxid, WatcherOrBitSet supress) {
WatchedEvent e = new WatchedEvent(type, KeeperState.SyncConnected, path, zxid);
Set<Watcher> watchers = new HashSet<>();
synchronized (this) {
// path迭代器，从子节点path向前遍历
// 例如/apps/app1/name
// next = /apps/app1/name, next = /apps/app1, next = /apps ...
PathParentIterator pathParentIterator = getPathParentIterator(path);
for (String localPath : pathParentIterator.asIterable()) {
// 获取遍历Watcher集
Set<Watcher> thisWatchers = watchTable.get(localPath);
Iterator<Watcher> iterator = thisWatchers.iterator();
while (iterator.hasNext()) {
Watcher watcher = iterator.next();
// 获取watcher对应的WatchStats
Map<String, WatchStats> paths = watch2Paths.getOrDefault(watcher, Collections.emptyMap());
WatchStats stats = paths.get(localPath); // if stats==null continue
if (!pathParentIterator.atParentPath()) {
watchers.add(watcher); // 加入watchers中
WatchStats newStats = stats.removeMode(WatcherMode.STANDARD);
if (newStats == WatchStats.NONE) { // STANDARD模式下会移除监听器
iterator.remove();
paths.remove(localPath);
} else if (newStats != stats) {
paths.put(localPath, newStats);
}
} else if (stats.hasMode(WatcherMode.PERSISTENT_RECURSIVE)) {
// 递归模式下才将父节点加入watchers中
watchers.add(watcher);
}
}
if (thisWatchers.isEmpty()) {
watchTable.remove(localPath);
}
}
}
// 略
for (Watcher w : watchers) {
if (supress != null && supress.contains(w)) {
continue;
}
w.process(e);
}
// 略
return new WatcherOrBitSet(watchers);
}
NIOServerCnxn
上面查找到watchers之后会触发process方法，看一下NIOServerCnxn的方法实现：
public void process(WatchedEvent event) {
ReplyHeader h = new ReplyHeader(ClientCnxn.NOTIFICATION_XID, event.getZxid(), 0);
// 转型成WatcherEvent才能通过网络传输
WatcherEvent e = event.getWrapper();
// 把事件推送给客户端
int responseSize = sendResponse(h, e, "notification", null, null, ZooDefs.OpCode.error);
ServerMetrics.getMetrics().WATCH_BYTES.add(responseSize);
}
