zookeeper源码(06)ZooKeeperServer及子类

ZooKeeperServer

实现了单机版zookeeper服务端功能，子类实现了更加丰富的分布式集群功能：

1. ZooKeeperServer
2.   |-- QuorumZooKeeperServer
3.     |-- LeaderZooKeeperServer
4.     |-- LearnerZooKeeperServer
5.       |-- FollowerZooKeeperServer
6.       |-- ObserverZooKeeperServer
7.   |-- ReadOnlyZooKeeperServer

复制代码

主要字段

1. // tickTime参数默认值
2. public static final int DEFAULT_TICK_TIME = 3000;
3. protected int tickTime = DEFAULT_TICK_TIME;
4. // 默认tickTime * 2
5. protected int minSessionTimeout = -1;
6. // 默认tickTime * 20
7. protected int maxSessionTimeout = -1;
9. // 会话跟踪
10. protected SessionTracker sessionTracker;
12. // 存储组件
13. private FileTxnSnapLog txnLogFactory = null;
14. private ZKDatabase zkDb;
16. // 缓存数据
17. private ResponseCache readResponseCache;
18. private ResponseCache getChildrenResponseCache;
20. // zxid会在启动阶段设置为最新lastZxid
21. private final AtomicLong hzxid = new AtomicLong(0);
22. // 请求处理器链入口
23. protected RequestProcessor firstProcessor;
24. // 缓存变化的数据
25. final Deque<ChangeRecord> outstandingChanges = new ArrayDeque<>();
26. final Map<String, ChangeRecord> outstandingChangesForPath = new HashMap<>();
28. protected ServerCnxnFactory serverCnxnFactory;
29. protected ServerCnxnFactory secureServerCnxnFactory;
31. // 大请求判断使用的参数
32. private volatile int largeRequestMaxBytes = 100 * 1024 * 1024;
33. private volatile int largeRequestThreshold = -1;

复制代码

主要方法

方法定义

1. // 通过zkDb从dataTree中删除Watcher监听器
2. void removeCnxn(ServerCnxn cnxn);
4. // 创建zkDb(为null时)并loadData加载数据
5. public void startdata() throws IOException, InterruptedException;
6. // 加载数据、清理session、生成快照
7. public void loadData() throws IOException, InterruptedException;
8. // 保存zkDb当前快照
9. public synchronized File takeSnapshot(boolean syncSnap, boolean isSevere,
10.                                       boolean fastForwardFromEdits) throws IOException;
12. // 从指定的输入流解析数据，生成新的zkDb和SessionTrack
13. public synchronized long restoreFromSnapshot(final InputStream inputStream) throws IOException;
15. // 使用zkDb.truncateLog(zxid)删除快照数据
16. public void truncateLog(long zxid) throws IOException;
17. // 通过zkDb获取dataTree.lastProcessedZxid的值
18. public long getLastProcessedZxid();
20. // 提交closeSession类型的Request来关闭会话
21. private void close(long sessionId);
22. // 使用zkDb杀掉会话
23. protected void killSession(long sessionId, long zxid);
24. // 启动组件
25. private void startupWithServerState(State state);
26. // 创建RequestProcessor用来处理请求
27. protected void setupRequestProcessors();
28. // 创建SessionTracker
29. protected void createSessionTracker();
31. // 为指定的session生成一个密码
32. byte[] generatePasswd(long id);
33. // 验证session密码
34. protected boolean checkPasswd(long sessionId, byte[] passwd);
35. // 使用sessionTracker创建session、生成密码、提交一个createSession请求
36. long createSession(ServerCnxn cnxn, byte[] passwd, int timeout);
37. // 为指定的session绑定owner
38. public void setOwner(long id, Object owner) throws SessionExpiredException;
39. // 验证session之后使用finishSessionInit方法确定继续通信或者断开连接
40. protected void revalidateSession(ServerCnxn cnxn, long sessionId, int sessionTimeout) throws IOException;
41. public void finishSessionInit(ServerCnxn cnxn, boolean valid);
42. // checkPasswd->revalidateSession->finishSessionInit
43. public void reopenSession(ServerCnxn cnxn, long sessionId,
44.                           byte[] passwd, int sessionTimeout) throws IOException;
46. // 把请求提交给requestThrottler之后再陆续调用submitRequestNow处理
47. public void enqueueRequest(Request si);
48. // 使用firstProcessor处理请求
49. public void submitRequestNow(Request si);
51. // 处理连接请求，网络IO层调用
52. public void processConnectRequest(
53.     ServerCnxn cnxn, ConnectRequest request) throws IOException, ClientCnxnLimitException;
54. // 处理业务请求，网络IO层调用
55. public void processPacket(ServerCnxn cnxn, RequestHeader h, RequestRecord request) throws IOException;
56. // sasl认证
57. private void processSasl(
58.     RequestRecord request, ServerCnxn cnxn, RequestHeader requestHeader) throws IOException;
60. // 处理transaction
61. public ProcessTxnResult processTxn(TxnHeader hdr, Record txn);
62. public ProcessTxnResult processTxn(Request request);
63. private void processTxnForSessionEvents(Request request, TxnHeader hdr, Record txn);
64. private ProcessTxnResult processTxnInDB(TxnHeader hdr, Record txn, TxnDigest digest);
66. // Grant or deny authorization to an operation on a node
67. public void checkACL(ServerCnxn cnxn, List<ACL> acl, int perm, List<Id> ids,
68.                      String path, List<ACL> setAcls) throws KeeperException.NoAuthException;
69. // Check a path whether exceeded the quota
70. public void checkQuota(String path, byte[] lastData, byte[] data,
71.                        int type) throws KeeperException.QuotaExceededException;
72. private void checkQuota(String lastPrefix, long bytesDiff, long countDiff,
73.                         String namespace) throws KeeperException.QuotaExceededException;
75. // 获取上级父类path
76. private String parentPath(String path) throws KeeperException.BadArgumentsException;
77. // 从Request获取有效的path
78. private String effectiveACLPath(
79.     Request request) throws KeeperException.BadArgumentsException, KeeperException.InvalidACLException;
80. // 根据Request获取需要的权限类型
81. private int effectiveACLPerms(Request request);
82. // 检查写权限
83. public boolean authWriteRequest(Request request);

复制代码

loadData方法

加载数据、清理session、生成快照：

1. public void loadData() throws IOException, InterruptedException {
2.     // 初始化zxid
3.     if (zkDb.isInitialized()) {
4.         setZxid(zkDb.getDataTreeLastProcessedZxid());
5.     } else {
6.         setZxid(zkDb.loadDataBase());
7.     }
9.     // 使用killSession方法杀死过期会话
10.     zkDb.getSessions().stream()
11.                     .filter(session -> zkDb.getSessionWithTimeOuts().get(session) == null)
12.                     .forEach(session -> killSession(session, zkDb.getDataTreeLastProcessedZxid()));
14.     // 保存快照
15.     // txnLogFactory.save(zkDb.getDataTree(), zkDb.getSessionWithTimeOuts(), syncSnap)
16.     takeSnapshot();
17. }

复制代码

killSession方法

1. protected void killSession(long sessionId, long zxid) {
2.     // 需要清理临时节点
3.     zkDb.killSession(sessionId, zxid);
4.     if (sessionTracker != null) {
5.         // 删除会话跟踪信息
6.         sessionTracker.removeSession(sessionId);
7.     }
8. }

复制代码

startupWithServerState方法

1. private void startupWithServerState(State state) {
2.     if (sessionTracker == null) {
3.         createSessionTracker();
4.     }
5.     startSessionTracker();
6.     // 创建RequestProcessor用于处理请求
7.     setupRequestProcessors();
9.     // 这是一个限流的组件，不做分析
10.     startRequestThrottler();
12.     registerJMX();
13.     startJvmPauseMonitor();
14.     registerMetrics();
16.     setState(state);
18.     requestPathMetricsCollector.start();
20.     localSessionEnabled = sessionTracker.isLocalSessionsEnabled();
22.     notifyAll();
23. }

复制代码

setupRequestProcessors方法(重要)

1. protected void setupRequestProcessors() {
2.     RequestProcessor finalProcessor = new FinalRequestProcessor(this);
3.     RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor);
4.     ((SyncRequestProcessor) syncProcessor).start();
5.     firstProcessor = new PrepRequestProcessor(this, syncProcessor);
6.     ((PrepRequestProcessor) firstProcessor).start();
7. }

复制代码

RequestProcessor接口：以处理器链方式处理事务，请求总是按顺序处理。standaloneServer、follower和leader有不同的处理器链。请求通过processRequest方法传递给其他RequestProcessor对象，通常情况总是由单个线程调用。当调用shutdown时，RequestProcessor还应关闭与其关联的其他RequestProcessor对象。
FinalRequestProcessor类：处理与请求相关的事务，并提供查询服务，给客户端发响应，位于RequestProcessor链末尾。
SyncRequestProcessor类：将请求记录到磁盘，对请求进行批处理，以有效地执行IO操作。在日志同步到磁盘之前，请求不会传递给下一个RequestProcessor对象。SyncRequestProcessor用于3种不同的情况：

• Leader - 将请求同步到磁盘，并将其转发给AckRequestProcessor，后者将ack发送回leader自己
  
• Follower - 将请求同步到磁盘，并将其转发给SendAckRequestProcessor，后者将ack发送给leader
  
• Observer - 将请求同步到磁盘，作为INFORM数据包接收。不将ack发送回leader，因此nextProcessor将为null
  

PrepRequestProcessor类：通常位于RequestProcessor链开头，为更新请求关联的事务做设置。
createSessionTracker方法

1. protected void createSessionTracker() {
2.     sessionTracker = new SessionTrackerImpl(this, zkDb.getSessionWithTimeOuts(), tickTime,
3.                                             createSessionTrackerServerId, getZooKeeperServerListener());
4. }

复制代码

不同的子类使用了不同的SessionTracker实现类：

• LeaderZooKeeperServer - LeaderSessionTracker
  
• LearnerZooKeeperServer- LearnerSessionTracker
  

createSession方法

1. long createSession(ServerCnxn cnxn, byte[] passwd, int timeout) {
2.     if (passwd == null) {
3.         passwd = new byte[0];
4.     }
5.     // 创建一个session
6.     long sessionId = sessionTracker.createSession(timeout);
7.     // 生成session密码
8.     Random r = new Random(sessionId ^ superSecret);
9.     r.nextBytes(passwd);
10.     // 提交createSession请求，该请求会被RequestProcessor处理
11.     CreateSessionTxn txn = new CreateSessionTxn(timeout);
12.     cnxn.setSessionId(sessionId);
13.     Request si = new Request(cnxn, sessionId, 0, OpCode.createSession, RequestRecord.fromRecord(txn), null);
14.     submitRequest(si);
15.     return sessionId;
16. }

复制代码

submitRequestNow方法

1. public void submitRequestNow(Request si) {
2.     try {
3.         touch(si.cnxn);
4.         boolean validpacket = Request.isValid(si.type);
5.         if (validpacket) {
6.             setLocalSessionFlag(si);
7.             // 使用firstProcessor处理请求
8.             firstProcessor.processRequest(si);
9.             if (si.cnxn != null) {
10.                 incInProcess();
11.             }
12.         } else {
13.             // Update request accounting/throttling limits
14.             requestFinished(si);
15.             new UnimplementedRequestProcessor().processRequest(si);
16.         }
17.     } catch (MissingSessionException e) {
18.         // Update request accounting/throttling limits
19.         requestFinished(si);
20.     } catch (RequestProcessorException e) {
21.         // Update request accounting/throttling limits
22.         requestFinished(si);
23.     }
24. }

复制代码

processConnectRequest方法

1. public void processConnectRequest(
2.         ServerCnxn cnxn, ConnectRequest request) throws IOException, ClientCnxnLimitException {
4.     long sessionId = request.getSessionId();
5.     // 略
7.     if (request.getLastZxidSeen() > zkDb.dataTree.lastProcessedZxid) {
8.         // zxid参数有误
9.         throw new CloseRequestException(msg, ServerCnxn.DisconnectReason.CLIENT_ZXID_AHEAD);
10.     }
11.     int sessionTimeout = request.getTimeOut();
12.     byte[] passwd = request.getPasswd();
13.     int minSessionTimeout = getMinSessionTimeout();
14.     if (sessionTimeout < minSessionTimeout) {
15.         sessionTimeout = minSessionTimeout;
16.     }
17.     int maxSessionTimeout = getMaxSessionTimeout();
18.     if (sessionTimeout > maxSessionTimeout) {
19.         sessionTimeout = maxSessionTimeout;
20.     }
21.     cnxn.setSessionTimeout(sessionTimeout);
22.     // We don't want to receive any packets until we are sure that the session is setup
23.     cnxn.disableRecv();
24.     if (sessionId == 0) {
25.         // 创建session
26.         long id = createSession(cnxn, passwd, sessionTimeout);
27.     } else {
28.         validateSession(cnxn, sessionId); // do nothing
29.         // 关闭旧的ServerCnxn
30.         if (serverCnxnFactory != null) {
31.             serverCnxnFactory.closeSession(sessionId, ServerCnxn.DisconnectReason.CLIENT_RECONNECT);
32.         }
33.         if (secureServerCnxnFactory != null) {
34.             secureServerCnxnFactory.closeSession(sessionId, ServerCnxn.DisconnectReason.CLIENT_RECONNECT);
35.         }
36.         cnxn.setSessionId(sessionId);
37.         // 开启新session
38.         reopenSession(cnxn, sessionId, passwd, sessionTimeout);
39.     }
40. }

复制代码

processPacket方法

1. public void processPacket(ServerCnxn cnxn, RequestHeader h, RequestRecord request) throws IOException {
2.     cnxn.incrOutstandingAndCheckThrottle(h);
4.     if (h.getType() == OpCode.auth) {
5.         AuthPacket authPacket = request.readRecord(AuthPacket::new);
6.         String scheme = authPacket.getScheme();
7.         ServerAuthenticationProvider ap = ProviderRegistry.getServerProvider(scheme);
8.         Code authReturn = KeeperException.Code.AUTHFAILED;
9.         // 认证、继续通信或者关闭连接，略
10.         return;
11.     } else if (h.getType() == OpCode.sasl) {
12.         processSasl(request, cnxn, h);
13.     } else {
14.         if (!authHelper.enforceAuthentication(cnxn, h.getXid())) {
15.             return;
16.         } else {
17.             Request si = new Request(
18.                 cnxn, cnxn.getSessionId(), h.getXid(), h.getType(), request, cnxn.getAuthInfo());
19.             int length = request.limit();
20.             if (isLargeRequest(length)) { // 判断large请求
21.                 checkRequestSizeWhenMessageReceived(length);
22.                 si.setLargeRequestSize(length);
23.             }
24.             si.setOwner(ServerCnxn.me);
25.             // 提交请求等待firstProcessor处理
26.             submitRequest(si);
27.         }
28.     }
29. }

复制代码

processTxn相关方法

1. // entry point for quorum/Learner.java
2. public ProcessTxnResult processTxn(TxnHeader hdr, Record txn) {
3.     processTxnForSessionEvents(null, hdr, txn);
4.     return processTxnInDB(hdr, txn, null);
5. }
7. // entry point for FinalRequestProcessor.java
8. public ProcessTxnResult processTxn(Request request) {
9.     TxnHeader hdr = request.getHdr();
10.     processTxnForSessionEvents(request, hdr, request.getTxn());
12.     final boolean writeRequest = (hdr != null);
13.     final boolean quorumRequest = request.isQuorum();
15.     // return fast w/o synchronization when we get a read
16.     if (!writeRequest && !quorumRequest) {
17.         return new ProcessTxnResult();
18.     }
19.     synchronized (outstandingChanges) {
20.         ProcessTxnResult rc = processTxnInDB(hdr, request.getTxn(), request.getTxnDigest());
22.         // request.hdr is set for write requests, which are the only ones
23.         // that add to outstandingChanges.
24.         if (writeRequest) {
25.             long zxid = hdr.getZxid();
26.             while (!outstandingChanges.isEmpty() && outstandingChanges.peek().zxid <= zxid) {
27.                 ChangeRecord cr = outstandingChanges.remove();
28.                 ServerMetrics.getMetrics().OUTSTANDING_CHANGES_REMOVED.add(1);
29.                 if (outstandingChangesForPath.get(cr.path) == cr) {
30.                     outstandingChangesForPath.remove(cr.path);
31.                 }
32.             }
33.         }
35.         // do not add non quorum packets to the queue.
36.         if (quorumRequest) {
37.             getZKDatabase().addCommittedProposal(request);
38.         }
39.         return rc;
40.     }
41. }
43. private void processTxnForSessionEvents(Request request, TxnHeader hdr, Record txn) {
44.     int opCode = (request == null) ? hdr.getType() : request.type;
45.     long sessionId = (request == null) ? hdr.getClientId() : request.sessionId;
47.     if (opCode == OpCode.createSession) {
48.         if (hdr != null && txn instanceof CreateSessionTxn) {
49.             CreateSessionTxn cst = (CreateSessionTxn) txn;
50.             // Add the session to the local session map or global one in zkDB.
51.             sessionTracker.commitSession(sessionId, cst.getTimeOut());
52.         }
53.     } else if (opCode == OpCode.closeSession) {
54.         sessionTracker.removeSession(sessionId);
55.     }
56. }
58. private ProcessTxnResult processTxnInDB(TxnHeader hdr, Record txn, TxnDigest digest) {
59.     if (hdr == null) {
60.         return new ProcessTxnResult();
61.     } else {
62.         return getZKDatabase().processTxn(hdr, txn, digest);
63.     }
64. }

复制代码

实现类概述

集群模式下leader节点使用的ZooKeeperServer实现类：

• 继承QuorumZooKeeperServer
  
• 使用的RequestProcessor与父类不同：
  1. Just like the standard ZooKeeperServer. We just replace the request processors: PrepRequestProcessor -> ProposalRequestProcessor -> CommitProcessor -> Leader.ToBeAppliedRequestProcessor -> FinalRequestProcessor

  复制代码
• 使用LeaderSessionTracker做会话追踪
  
• 与learner节点通信
  

处理器链

• FinalRequestProcessor - 处理与请求相关的事务，并提供查询服务，给客户端发响应，位于RequestProcessor链末尾
  
• ToBeAppliedRequestProcessor - 维护toBeApplied列表
  
• CommitProcessor - 等待commit完成之后调用下游RequestProcessor处理器
  
• ProposalRequestProcessor - 发起proposal并将Request转发给内部的SyncRequestProcessor和AckRequestProcessor
  1. // 构建处理器链
  2. protected void setupRequestProcessors() {
  3.     RequestProcessor finalProcessor = new FinalRequestProcessor(this);
  4.     RequestProcessor toBeAppliedProcessor =
  5.         new Leader.ToBeAppliedRequestProcessor(finalProcessor, getLeader());
  6.     commitProcessor = new CommitProcessor(
  7.         toBeAppliedProcessor, Long.toString(getServerId()), false, getZooKeeperServerListener());
  8.     commitProcessor.start();
  9.     ProposalRequestProcessor proposalProcessor = new ProposalRequestProcessor(this, commitProcessor);
  10.     proposalProcessor.initialize();
  11.     prepRequestProcessor = new PrepRequestProcessor(this, proposalProcessor);
  12.     prepRequestProcessor.start();
  13.     firstProcessor = new LeaderRequestProcessor(this, prepRequestProcessor);
  15.     setupContainerManager(); // 启动ContainerManager用于删除ttl节点和container节点
  16. }

  复制代码
• PrepRequestProcessor - 通常位于RequestProcessor链开头，为更新请求关联的事务做设置
  
• LeaderRequestProcessor - 负责执行本地会话升级，只有直接提交给leader的Request才能通过这个处理器
  

LearnerZooKeeperServer

Learner基类：

• 使用LearnerSessionTracker做会话追踪
  
• 使用CommitProcessor、SyncRequestProcessor做处理器链
  

FollowerZooKeeperServer

实现类概述

与ZooKeeperServer类似，只是处理器链不同：
FollowerRequestProcessor -> CommitProcessor -> FinalRequestProcessor
使用SyncRequestProcessor来记录leader的提案。
处理器链

setupRequestProcessors方法：

1. public ProposalRequestProcessor(LeaderZooKeeperServer zks, RequestProcessor nextProcessor) {
2.     this.zks = zks;
3.     this.nextProcessor = nextProcessor;
4.     // 内部有维护SyncRequestProcessor和AckRequestProcessor
5.     AckRequestProcessor ackProcessor = new AckRequestProcessor(zks.getLeader());
6.     syncProcessor = new SyncRequestProcessor(zks, ackProcessor);
8.     forwardLearnerRequestsToCommitProcessorDisabled = Boolean.getBoolean(
9.             FORWARD_LEARNER_REQUESTS_TO_COMMIT_PROCESSOR_DISABLED);
10. }

复制代码

• FinalRequestProcessor
  
• CommitProcessor
  
• FollowerRequestProcessor - 将数据更新请求转发给Leader
  
• SyncRequestProcessor
  
• SendAckRequestProcessor - 给leader发ACK
  

ObserverZooKeeperServer

Observer类型节点的ZooKeeperServer实现。
setupRequestProcessors方法：

1. protected void setupRequestProcessors() {
2.     RequestProcessor finalProcessor = new FinalRequestProcessor(this);
3.     commitProcessor = new CommitProcessor(
4.         finalProcessor, Long.toString(getServerId()), true, getZooKeeperServerListener());
5.     commitProcessor.start();
6.     firstProcessor = new FollowerRequestProcessor(this, commitProcessor);
7.     ((FollowerRequestProcessor) firstProcessor).start();
8.     syncProcessor = new SyncRequestProcessor(this, new SendAckRequestProcessor(getFollower()));
9.     syncProcessor.start();
10. }

复制代码

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