zookeeper源码(10)node增删改查及监听

本文将从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事务准备

反序列化请求参数

1. switch (request.type) {
2. case OpCode.createContainer:
3. case OpCode.create:
4. case OpCode.create2:
5.     CreateRequest create2Request = request.readRequestRecord(CreateRequest::new);
6.     pRequest2Txn(request.type, zks.getNextZxid(), request, create2Request);
7.     break;
8. case OpCode.createTTL:
9.     // 默认不支持ttl
10.     CreateTTLRequest createTtlRequest = request.readRequestRecord(CreateTTLRequest::new);
11.     pRequest2Txn(request.type, zks.getNextZxid(), request, createTtlRequest);
12.     break;
13. // ...

复制代码

CreateRequest封装创建node的参数：

1. public class CreateRequest implements Record {
2.   private String path;
3.   private byte[] data;
4.   private java.util.List<org.apache.zookeeper.data.ACL> acl;
5.   private int flags;
6. }

复制代码

CreateTTLRequest封装创建node加ttl的参数：

1. public class CreateTTLRequest implements Record {
2.   private String path;
3.   private byte[] data;
4.   private java.util.List<org.apache.zookeeper.data.ACL> acl;
5.   private int flags;
6.   private long ttl;
7. }

复制代码

事务准备

1. protected void pRequest2Txn(int type, long zxid, Request request, Record record)
2.         throws KeeperException, IOException, RequestProcessorException {
3.     // ...
5.     switch (type) {
6.       case OpCode.create:
7.       case OpCode.create2:
8.       case OpCode.createTTL:
9.       case OpCode.createContainer: {
10.         pRequest2TxnCreate(type, request, record);
11.         break;
12.       }
13.     // ...
14.     }
15. }
17. private void pRequest2TxnCreate(
18.         int type, Request request, Record record) throws IOException, KeeperException {
19.     int flags;
20.     String path;
21.     List<ACL> acl;
22.     byte[] data;
23.     long ttl;
24.     if (type == OpCode.createTTL) {
25.         CreateTTLRequest createTtlRequest = (CreateTTLRequest) record;
26.         // 给flags等参数赋值
27.     } else {
28.         CreateRequest createRequest = (CreateRequest) record;
29.         // 给flags等参数赋值
30.         ttl = -1;
31.     }
32.     // CreateMode:
33.     // PERSISTENT, PERSISTENT_SEQUENTIAL, EPHEMERAL, EPHEMERAL_SEQUENTIAL,
34.     // CONTAINER, PERSISTENT_WITH_TTL, PERSISTENT_SEQUENTIAL_WITH_TTL
35.     CreateMode createMode = CreateMode.fromFlag(flags);
36.     // 验证临时节点、ttl参数、检查session
37.     // 默认不支持ttl
38.     validateCreateRequest(path, createMode, request, ttl);
39.     String parentPath = validatePathForCreate(path, request.sessionId); // 父节点path
41.     List<ACL> listACL = fixupACL(path, request.authInfo, acl); // 请求携带的权限
42.     ChangeRecord parentRecord = getRecordForPath(parentPath); // 得到父节点
43.     // 验证CREATE权限
44.     zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.CREATE, request.authInfo, path, listACL);
45.     int parentCVersion = parentRecord.stat.getCversion();
46.     if (createMode.isSequential()) { // 顺序节点
47.         // 例如/users/admin0000000001
48.         path = path + String.format(Locale.ENGLISH, "%010d", parentCVersion);
49.     }
50.     validatePath(path, request.sessionId);
51.     // 略
52.     boolean ephemeralParent =
53.         EphemeralType.get(parentRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL;
54.     // 父节点不可以是临时节点
56.     int newCversion = parentRecord.stat.getCversion() + 1; // 父节点的childVersion++
57.     // 检查字节限额
58.     zks.checkQuota(path, null, data, OpCode.create);
59.     // 不同类型创建对应的Txn对象
60.     if (type == OpCode.createContainer) {
61.         request.setTxn(new CreateContainerTxn(path, data, listACL, newCversion));
62.     } else if (type == OpCode.createTTL) {
63.         request.setTxn(new CreateTTLTxn(path, data, listACL, newCversion, ttl));
64.     } else {
65.         request.setTxn(new CreateTxn(path, data, listACL, createMode.isEphemeral(), newCversion));
66.     }
68.     TxnHeader hdr = request.getHdr();
69.     long ephemeralOwner = 0;
70.     if (createMode.isContainer()) {
71.         ephemeralOwner = EphemeralType.CONTAINER_EPHEMERAL_OWNER;
72.     } else if (createMode.isTTL()) {
73.         ephemeralOwner = EphemeralType.TTL.toEphemeralOwner(ttl);
74.     } else if (createMode.isEphemeral()) {
75.         ephemeralOwner = request.sessionId; // 临时节点使用sessionId
76.     }
77.     // czxid(created),mzxid(modified),ctime,mtime,version,cversion(childVersion),
78.     // aversion(aclVersion),ephemeralOwner,pzxid(lastModifiedChildren)
79.     StatPersisted s = DataTree.createStat(hdr.getZxid(), hdr.getTime(), ephemeralOwner);
81.     // 父节点
82.     parentRecord = parentRecord.duplicate(request.getHdr().getZxid());
83.     parentRecord.childCount++;
84.     parentRecord.stat.setCversion(newCversion);
85.     parentRecord.stat.setPzxid(request.getHdr().getZxid());
86.     parentRecord.precalculatedDigest = precalculateDigest(
87.             DigestOpCode.UPDATE, parentPath, parentRecord.data, parentRecord.stat);
88.     addChangeRecord(parentRecord);
89.     // 新增节点
90.     ChangeRecord nodeRecord = new ChangeRecord(request.getHdr().getZxid(), path, s, 0, listACL);
91.     nodeRecord.data = data;
92.     nodeRecord.precalculatedDigest = precalculateDigest(DigestOpCode.ADD, path, nodeRecord.data, s);
93.     setTxnDigest(request, nodeRecord.precalculatedDigest);
94.     addChangeRecord(nodeRecord);
95. }
97. protected void addChangeRecord(ChangeRecord c) {
98.     synchronized (zks.outstandingChanges) {
99.         zks.outstandingChanges.add(c);
100.         zks.outstandingChangesForPath.put(c.path, c);
101.     }
102. }

复制代码

outstandingChanges生存未提交的事务变化，比如在天生顺序节点时需要使用cversion值，但是在事务提交到ZKDatabase之前，库里面的值是旧的，所以在上面的代码中，是从outstandingChanges查找节点，给cversion++后再天生顺序节点。
在事务提交之后，才会清算outstandingChanges集。
ProposalRequestProcessor发Proposal

1. public void processRequest(Request request) throws RequestProcessorException {
2.     if (request instanceof LearnerSyncRequest) { // sync命令流程，暂不分析
3.         zks.getLeader().processSync((LearnerSyncRequest) request);
4.     } else {
5.         if (shouldForwardToNextProcessor(request)) {
6.             nextProcessor.processRequest(request); // 提交给下游处理器
7.         }
8.         if (request.getHdr() != null) { // 事务操作需要发proposal并写磁盘
9.             try {
10.                 zks.getLeader().propose(request);
11.             } catch (XidRolloverException e) {
12.                 throw new RequestProcessorException(e.getMessage(), e);
13.             }
14.             // 把事务log写到文件中
15.             // 之后通过AckRequestProcessor处理器给leader ack
16.             syncProcessor.processRequest(request);
17.         }
18.     }
19. }

复制代码

CommitProcessor提交事务

1. public void processRequest(Request request) {
2.     request.commitProcQueueStartTime = Time.currentElapsedTime();
3.     queuedRequests.add(request); // 所有请求队列
4.     if (needCommit(request)) { // 需要提交的请求进入到写队列
5.         queuedWriteRequests.add(request);
6.         numWriteQueuedRequests.incrementAndGet();
7.     } else {
8.         numReadQueuedRequests.incrementAndGet();
9.     }
10.     wakeup();
11. }

复制代码

run方法对比queuedRequests、queuedWriteRequests、committedRequests这几个队列，将提交成功的请求或读请求转发给鄙俚的ToBeAppliedRequestProcessor处置惩罚器。
FinalRequestProcessor应用事务

该处置惩罚器位于处置惩罚器链的末了，负责将事务应用到ZKDatabase、查询数据、返反相应。
applyRequest

该方法将事务应用到ZKDatabase中：

1. private ProcessTxnResult applyRequest(Request request) {
2.     // 应用事务
3.     ProcessTxnResult rc = zks.processTxn(request);
5.     // closeSession
7.     // metrics
9.     return rc;
10. }

复制代码

zks.processTxn负责处置惩罚session、处置惩罚事务、清算outstandingChanges集。重点看一下处置惩罚事务的步调。
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(
16.                 createTxn.getPath(),
17.                 createTxn.getData(),
18.                 createTxn.getAcl(),
19.                 createTxn.getEphemeral() ? header.getClientId() : 0,
20.                 createTxn.getParentCVersion(),
21.                 header.getZxid(),
22.                 header.getTime(),
23.                 null);
24.             break;
25.         case OpCode.create2:
26.             CreateTxn create2Txn = (CreateTxn) txn;
27.             rc.path = create2Txn.getPath();
28.             Stat stat = new Stat();
29.             createNode(
30.                 create2Txn.getPath(),
31.                 create2Txn.getData(),
32.                 create2Txn.getAcl(),
33.                 create2Txn.getEphemeral() ? header.getClientId() : 0,
34.                 create2Txn.getParentCVersion(),
35.                 header.getZxid(),
36.                 header.getTime(),
37.                 stat);
38.             rc.stat = stat;
39.             break;
40.         case OpCode.createTTL:
41.             CreateTTLTxn createTtlTxn = (CreateTTLTxn) txn;
42.             rc.path = createTtlTxn.getPath();
43.             stat = new Stat();
44.             createNode(
45.                 createTtlTxn.getPath(),
46.                 createTtlTxn.getData(),
47.                 createTtlTxn.getAcl(),
48.                 EphemeralType.TTL.toEphemeralOwner(createTtlTxn.getTtl()),
49.                 createTtlTxn.getParentCVersion(),
50.                 header.getZxid(),
51.                 header.getTime(),
52.                 stat);
53.             rc.stat = stat;
54.             break;
55.         case OpCode.createContainer:
56.             CreateContainerTxn createContainerTxn = (CreateContainerTxn) txn;
57.             rc.path = createContainerTxn.getPath();
58.             stat = new Stat();
59.             createNode(
60.                 createContainerTxn.getPath(),
61.                 createContainerTxn.getData(),
62.                 createContainerTxn.getAcl(),
63.                 EphemeralType.CONTAINER_EPHEMERAL_OWNER,
64.                 createContainerTxn.getParentCVersion(),
65.                 header.getZxid(),
66.                 header.getTime(),
67.                 stat);
68.             rc.stat = stat;
69.             break;
70.         // ...
71.         }
72.     }
73.     // ...
74. }

复制代码

createNode

1. public void createNode(final String path, byte[] data, List<ACL> acl, long ephemeralOwner, int parentCVersion, long zxid, long time, Stat outputStat) throws NoNodeException, NodeExistsException {
2.     int lastSlash = path.lastIndexOf('/');
3.     String parentName = path.substring(0, lastSlash);
4.     String childName = path.substring(lastSlash + 1);
5.     StatPersisted stat = createStat(zxid, time, ephemeralOwner);
6.     DataNode parent = nodes.get(parentName); // 父节点必须存在
7.     if (parent == null) {
8.         throw new NoNodeException();
9.     }
10.     synchronized (parent) {
11.         Long acls = aclCache.convertAcls(acl);
13.         Set<String> children = parent.getChildren();
14.         if (children.contains(childName)) { // 节点不能存在
15.             throw new NodeExistsException();
16.         }
18.         nodes.preChange(parentName, parent);
19.         if (parentCVersion == -1) { // childVersion++
20.             parentCVersion = parent.stat.getCversion();
21.             parentCVersion++;
22.         }
24.         if (parentCVersion > parent.stat.getCversion()) {
25.             parent.stat.setCversion(parentCVersion);
26.             parent.stat.setPzxid(zxid);
27.         }
28.         // 创建node
29.         DataNode child = new DataNode(data, acls, stat);
30.         parent.addChild(childName);
31.         nodes.postChange(parentName, parent);
32.         nodeDataSize.addAndGet(getNodeSize(path, child.data));
33.         nodes.put(path, child); // 维护NodeHashMap
34.         // 处理临时节点
35.         EphemeralType ephemeralType = EphemeralType.get(ephemeralOwner);
36.         if (ephemeralType == EphemeralType.CONTAINER) {
37.             containers.add(path);
38.         } else if (ephemeralType == EphemeralType.TTL) {
39.             ttls.add(path);
40.         } else if (ephemeralOwner != 0) {
41.             HashSet<String> list = ephemerals.computeIfAbsent(ephemeralOwner, k -> new HashSet<>());
42.             synchronized (list) {
43.                 list.add(path);
44.             }
45.         }
46.         // 返回节点stat
47.         if (outputStat != null) {
48.             child.copyStat(outputStat);
49.         }
50.     }
51.     // now check if its one of the zookeeper node child 略
53.     // 触发NodeCreated监听
54.     dataWatches.triggerWatch(path, Event.EventType.NodeCreated, zxid);
55.     // 触发父节点的NodeChildrenChanged监听
56.     childWatches.triggerWatch(
57.         parentName.equals("") ? "/" : parentName, Event.EventType.NodeChildrenChanged, zxid);
58. }

复制代码

返反相应

1. case OpCode.create: {
2.     lastOp = "CREA";
3.     rsp = new CreateResponse(rc.path); // 创建Response
4.     err = Code.get(rc.err); // processTxn的err
5.     requestPathMetricsCollector.registerRequest(request.type, rc.path);
6.     break;
7. }
8. case OpCode.create2:
9. case OpCode.createTTL:
10. case OpCode.createContainer: {
11.     lastOp = "CREA";
12.     rsp = new Create2Response(rc.path, rc.stat); // 创建Response
13.     err = Code.get(rc.err); // processTxn的err
14.     requestPathMetricsCollector.registerRequest(request.type, rc.path);
15.     break;
16. }

复制代码

最后会使用cnxn把相应返回给客户端：

1. ReplyHeader hdr = new ReplyHeader(request.cxid, lastZxid, err.intValue());
2. 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事务准备

反序列化请求参数

1. private void pRequestHelper(Request request) {
2.     try {
3.         switch (request.type) {
4.         // ...
5.         case OpCode.deleteContainer:
6.             DeleteContainerRequest deleteContainerRequest =
7.                 request.readRequestRecord(DeleteContainerRequest::new);
8.             pRequest2Txn(request.type, zks.getNextZxid(), request, deleteContainerRequest);
9.             break;
10.         case OpCode.delete:
11.             DeleteRequest deleteRequest = request.readRequestRecord(DeleteRequest::new);
12.             pRequest2Txn(request.type, zks.getNextZxid(), request, deleteRequest);
13.             break;
14.         }
15.         // ...
16.     }
17. }

复制代码

DeleteContainerRequest类：

1. public class DeleteContainerRequest implements Record {
2.     private String path;
3. }

复制代码

DeleteRequest类：

1. public class DeleteRequest implements Record {
2.   private String path;
3.   private int version;
4. }

复制代码

事务准备

1. protected void pRequest2Txn(int type, long zxid, Request request,
2.              Record record) throws KeeperException, IOException, RequestProcessorException {
3.     // 略
5.     switch (type) {
6.     // 略
7.     case OpCode.deleteContainer: {
8.         DeleteContainerRequest txn = (DeleteContainerRequest) record;
9.         String path = txn.getPath();
10.         String parentPath = getParentPathAndValidate(path);
11.         ChangeRecord nodeRecord = getRecordForPath(path); // 获取待删除节点
12.         if (nodeRecord.childCount > 0) { // 有子节点不允许删除
13.             throw new KeeperException.NotEmptyException(path);
14.         }
15.         if (EphemeralType.get(nodeRecord.stat.getEphemeralOwner()) == EphemeralType.NORMAL) {
16.             throw new KeeperException.BadVersionException(path);
17.         }
18.         ChangeRecord parentRecord = getRecordForPath(parentPath); // 获取父节点
19.         request.setTxn(new DeleteTxn(path));
20.         // addChangeRecord 略
21.         break;
22.     }
23.     case OpCode.delete:
24.         zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
25.         DeleteRequest deleteRequest = (DeleteRequest) record;
26.         String path = deleteRequest.getPath();
27.         String parentPath = getParentPathAndValidate(path);
28.         ChangeRecord parentRecord = getRecordForPath(parentPath); // 获取父节点
29.         // 检查DELETE权限
30.         zks.checkACL(request.cnxn, parentRecord.acl, ZooDefs.Perms.DELETE, request.authInfo, path, null);
31.         ChangeRecord nodeRecord = getRecordForPath(path); // 获取待删除节点
32.         checkAndIncVersion(nodeRecord.stat.getVersion(), deleteRequest.getVersion(), path); // 检查version
33.         if (nodeRecord.childCount > 0) { // 有子节点不允许删除
34.             throw new KeeperException.NotEmptyException(path);
35.         }
36.         request.setTxn(new DeleteTxn(path));
37.         // addChangeRecord 略
38.         break;
39.     }
40. }

复制代码

FinalRequestProcessor应用事务

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.         // ...
13.         case OpCode.delete:
14.         case OpCode.deleteContainer:
15.             DeleteTxn deleteTxn = (DeleteTxn) txn;
16.             rc.path = deleteTxn.getPath();
17.             deleteNode(deleteTxn.getPath(), header.getZxid());
18.             break;
19.         }
20.         // ...
21.     }
22. }

复制代码

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.     if (parent == null) { // 获取父节点且必须存在
8.         throw new NoNodeException();
9.     }
10.     synchronized (parent) {
11.         nodes.preChange(parentName, parent);
12.         parent.removeChild(childName);
13.         if (zxid > parent.stat.getPzxid()) {
14.             parent.stat.setPzxid(zxid); // The zxid of the change that last modified children of this znode
15.         }
16.         nodes.postChange(parentName, parent);
17.     }
19.     DataNode node = nodes.get(path); // 获取删除节点
20.     if (node == null) {
21.         throw new NoNodeException();
22.     }
23.     nodes.remove(path); // 从NodeHashMap删除
24.     synchronized (node) { // 移除权限
25.         aclCache.removeUsage(node.acl);
26.         nodeDataSize.addAndGet(-getNodeSize(path, node.data));
27.     }
29.     // 移除临时节点、container、ttl等缓存
30.     synchronized (parent) {
31.         long owner = node.stat.getEphemeralOwner();
32.         EphemeralType ephemeralType = EphemeralType.get(owner);
33.         if (ephemeralType == EphemeralType.CONTAINER) {
34.             containers.remove(path);
35.         } else if (ephemeralType == EphemeralType.TTL) {
36.             ttls.remove(path);
37.         } else if (owner != 0) {
38.             Set<String> nodes = ephemerals.get(owner);
39.             if (nodes != null) {
40.                 synchronized (nodes) {
41.                     nodes.remove(path);
42.                 }
43.             }
44.         }
45.     }
47.     // 略
49.     // 触发NodeDeleted监听
50.     WatcherOrBitSet processed = dataWatches.triggerWatch(path, EventType.NodeDeleted, zxid);
51.     childWatches.triggerWatch(path, EventType.NodeDeleted, zxid, processed);
52.     // 触发父节点的NodeChildrenChanged监听
53.     childWatches.triggerWatch(
54.         "".equals(parentName) ? "/" : parentName, EventType.NodeChildrenChanged, zxid);
55. }

复制代码

设置node数据

PrepRequestProcessor事务准备

反序列化请求参数

1. private void pRequestHelper(Request request) {
2.     try {
3.         switch (request.type) {
4.         // ...
5.         case OpCode.setData:
6.             SetDataRequest setDataRequest = request.readRequestRecord(SetDataRequest::new);
7.             pRequest2Txn(request.type, zks.getNextZxid(), request, setDataRequest);
8.             break;
9.         // other case
10.         }
11.     }
12.     // ...
13. }

复制代码

SetDataRequest类：

1. public class SetDataRequest implements Record {
2.   private String path;
3.   private byte[] data;
4.   private int version;
5. }

复制代码

事务准备

1. protected void pRequest2Txn(int type, long zxid, Request request,
2.           Record record) throws KeeperException, IOException, RequestProcessorException {
3.     // 略
5.     switch (type) {
6.     // ...
7.     case OpCode.setData:
8.         zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
9.         SetDataRequest setDataRequest = (SetDataRequest) record;
10.         path = setDataRequest.getPath();
11.         validatePath(path, request.sessionId);
12.         nodeRecord = getRecordForPath(path); // 获取节点对象
13.         // 检查权限
14.         zks.checkACL(request.cnxn, nodeRecord.acl, ZooDefs.Perms.WRITE, request.authInfo, path, null);
15.         // 检查字节限额
16.         zks.checkQuota(path, nodeRecord.data, setDataRequest.getData(), OpCode.setData);
17.         // version++
18.         int newVersion = checkAndIncVersion(
19.             nodeRecord.stat.getVersion(), setDataRequest.getVersion(), path);
20.         // 创建SetDataTxn
21.         request.setTxn(new SetDataTxn(path, setDataRequest.getData(), newVersion));
22.         // addChangeRecord
23.         break;
24.     // other case
25.     }
26. }

复制代码

FinalRequestProcessor应用事务

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.         // other case
13.         case OpCode.setData:
14.             SetDataTxn setDataTxn = (SetDataTxn) txn;
15.             rc.path = setDataTxn.getPath();
16.             rc.stat = setData(
17.                 setDataTxn.getPath(),
18.                 setDataTxn.getData(),
19.                 setDataTxn.getVersion(),
20.                 header.getZxid(),
21.                 header.getTime());
22.             break;
23.         // other case
24.         }
25.     }
26.     // ...
27. }

复制代码

setData

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

复制代码

查询node数据

PrepRequestProcessor验证session

经过该处置惩罚器时，只做session验证。
之后的ProposalRequestProcessor、CommitProcessor、ToBeAppliedRequestProcessor都是直接通过，不做事务处置惩罚，直接交给FinalRequestProcessor处置惩罚器查询数据、发送相应。
FinalRequestProcessor查询数据

使用handleGetDataRequest方法查询数据：

1. private Record handleGetDataRequest(
2.         Record request, ServerCnxn cnxn, List<Id> authInfo) throws KeeperException, IOException {
3.     GetDataRequest getDataRequest = (GetDataRequest) request;
4.     String path = getDataRequest.getPath();
5.     DataNode n = zks.getZKDatabase().getNode(path);
6.     if (n == null) {
7.         throw new KeeperException.NoNodeException();
8.     }
9.     // 检查权限
10.     zks.checkACL(cnxn, zks.getZKDatabase().aclForNode(n), ZooDefs.Perms.READ, authInfo, path, null);
11.     Stat stat = new Stat();
12.     // 查询数据
13.     // 如果watcher参数不为null会给path添加一个监听器
14.     byte[] b = zks.getZKDatabase().getData(path, stat, getDataRequest.getWatch() ? cnxn : null);
15.     return new GetDataResponse(b, stat);
16. }

复制代码

GetDataRequest类：

1. public class GetDataRequest implements Record {
2.   private String path;
3.   private boolean watch;
4. }

复制代码

节点监听

addWatch下令

1. case OpCode.addWatch: {
2.     lastOp = "ADDW";
3.     AddWatchRequest addWatcherRequest = request.readRequestRecord(AddWatchRequest::new);
4.     // 最终使用DataTree的addWatch方法注册监听器
5.     // cnxn是ServerCnxn对象，实现了Watcher接口
6.     zks.getZKDatabase().addWatch(addWatcherRequest.getPath(), cnxn, addWatcherRequest.getMode());
7.     rsp = new ErrorResponse(0);
8.     break;
9. }

复制代码

DataTree的addWatch方法

1. public void addWatch(String basePath, Watcher watcher, int mode) {
2.     // PERSISTENT|PERSISTENT_RECURSIVE
3.     WatcherMode watcherMode = WatcherMode.fromZooDef(mode);
4.     // dataWatches和childWatches是WatchManager类型对象
5.     dataWatches.addWatch(basePath, watcher, watcherMode);
6.     if (watcherMode != WatcherMode.PERSISTENT_RECURSIVE) {
7.         childWatches.addWatch(basePath, watcher, watcherMode);
8.     }
9. }

复制代码

WatcherMode枚举

1. public enum WatcherMode {
2.     STANDARD(false, false),
3.     PERSISTENT(true, false), // persistent=0
4.     PERSISTENT_RECURSIVE(true, true), // persistentRecursive=1
5.     ;
6. }

复制代码

PERSISTENT和PERSISTENT_RECURSIVE是3.6.0版本新增的特性。
Watcher接口

实现类需要实现process方法：

1. void process(WatchedEvent event);

复制代码

WatchedEvent代表一个监听变乱：

1. public class WatchedEvent {
2.     // 当前zk服务器的状态
3.     private final KeeperState keeperState;
4.     // NodeCreated|NodeDeleted|NodeDataChanged|NodeChildrenChanged等
5.     private final EventType eventType;
6.     private final String path;
7.     private final long zxid;
8. }

复制代码

重要的实现类：

• 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.
核心字段：

1. // path -> Watcher集
2. private final Map<String, Set<Watcher>> watchTable = new HashMap<>();
3. // Watcher -> path->WatchStats(PERSISTENT|STANDARD + PERSISTENT|PERSISTENT_RECURSIVE等)
4. private final Map<Watcher, Map<String, WatchStats>> watch2Paths = new HashMap<>();

复制代码

WatchStats类

[code]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