zk源码—6.Leader选举的实现原理

大纲
1.zk是如何实现数据一致性的
(1)数据一致性分析
(2)实现数据一致性的广播模式
(3)实现数据一致性的恢复模式
2.zk是如何进行Leader选举的
(1)服务器启动时的Leader选举
(2)服务器运行时的Leader选举
(3)Leader选举的算法设计
(4)Leader选举的实现细节
 
1.zk是如何实现数据一致性的
(1)数据一致性分析
(2)实现数据一致性的广播模式
(3)实现数据一致性的恢复模式
 
zk集群中的服务器分为Leader服务器、Follower服务器及Observer服务器。Leader选举是一个过程，在这个过程中主要做了两项工作：
工作一：选举出Leader服务器
工作二：进行数据同步
 
zk中实现的一致性不是强一致性，而是最终一致性。即集群中各个服务器上的数据并不是每时每刻都保持一致的，而是即经过一段时间后，集群服务器上的数据才最终保持一致。
 
Leader服务器主要负责处理事务请求，当Leader服务器接收到客户端的事务请求时，会先向集群中的各机器针对该请求的提议发起投票询问。
 
(1)数据一致性分析
zk在集群中采取的是多数原则的方式来保证数据一致性。即当一个事务请求导致服务器上的数据发生改变时，只要保证多数机器的数据都正确变更了，就可保证系统数据一致性。
 
因为每个Follower服务器都可以看作是Leader服务器的数据副本，所以只要保证集群中大多数机器数据是一致的，那么在集群中个别机器出现故障时，zk集群依然能保证稳定运行。
 
(2)实现数据一致性的广播模式
一.首先Leader启动时会创建网络连接管理器LearnerCnxAcceptor等待Learner的连接
LearnerCnxAcceptor监听到Learner发起的连接后，会新建一个LearnerHandler实例专门负责Leader和该Learner之间的连接。启动LearnerHandler时，又会开启一个线程专门负责发送消息给Learner。如果Learner发生故障，那么Leader中为该Learner维护的LearnerHandler的ping()方法会检测到然后关闭相关线程和实例。

1. public class Leader {
2.     ...
3.     private final ServerSocket ss;
4.    
5.     Leader(QuorumPeer self, LeaderZooKeeperServer zk) throws IOException {
6.         ...
7.         //创建BIO的ServerSocket，监听端口，等待客户端发起连接
8.         ss = new ServerSocket();
9.         ss.bind(self.getQuorumAddress());
10.         ...
11.     }
12.    
13.     void lead() throws IOException, InterruptedException {
14.         ...
15.         cnxAcceptor = new LearnerCnxAcceptor();//网络连接器
16.         cnxAcceptor.start();
17.         ...
18.         while (true) {
19.             ...
20.             for (LearnerHandler f : getLearners()) {
21.                 //Leader向Learner发出心跳检测
22.                 f.ping();
23.             }
24.             ...
25.         }
26.     }
27.    
28.     class LearnerCnxAcceptor extends ZooKeeperCriticalThread {
29.         ...
30.         public void run() {
31.             ...
32.             while (!stop) {
33.                 //监听到客户端发起的连接，新建一个线程LearnerHandler专门进行处理
34.                 Socket s = ss.accept();
35.                 s.setSoTimeout(self.tickTime * self.initLimit);
36.                 s.setTcpNoDelay(nodelay);
37.                 BufferedInputStream is = new BufferedInputStream(s.getInputStream());
38.                 LearnerHandler fh = new LearnerHandler(s, is, Leader.this);
39.                 fh.start();
40.             }
41.             ...
42.         }
43.         ...
44.     }
45.     ...
46. }
48. public class LearnerHandler extends ZooKeeperThread {
49.     ...
50.     public void run() {
51.         //先进行数据同步
52.         ...
53.         //开启线程发送信息给Learner
54.         startSendingPackets();
55.         ...
56.         //处理Learner发过来的消息，比如投票响应ACK消息、心跳响应PING消息等
57.         while (true) {
58.             qp = new QuorumPacket();
59.             ia.readRecord(qp, "packet");
60.             ...
61.         }
62.     }
63.    
64.     protected void startSendingPackets() {
65.         if (!sendingThreadStarted) {
66.             new Thread() {
67.                 public void run() {
68.                     Thread.currentThread().setName("Sender-" + sock.getRemoteSocketAddress());
69.                     try {
70.                         sendPackets();
71.                     } catch (InterruptedException e) {
72.                         LOG.warn("Unexpected interruption " + e.getMessage());
73.                     }
74.                 }
75.             }.start();
76.             sendingThreadStarted = true;
77.         } else {
78.             LOG.error("Attempting to start sending thread after it already started");
79.         }
80.     }
81.    
82.     private void sendPackets() throws InterruptedException {
83.         long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
84.         while (true) {
85.             try {
86.                 //从queuedPackets队列中提取消息出来发送给Learner
87.                 QuorumPacket p = queuedPackets.poll();
88.                 if (p == proposalOfDeath) {
89.                     // Packet of death!
90.                     break;
91.                 }
92.                 oa.writeRecord(p, "packet");
93.             } catch (IOException e) {
94.                 //假如Leader在这里向Learner发送消息时，Learner故障了，那么就会在这里报错
95.                 //此时，这里的报错并不影响对应的LearnerHandler实例和Leader实例
96.                 if (!sock.isClosed()) {
97.                     LOG.warn("Unexpected exception at " + this, e);
98.                     try {
99.                         sock.close();
100.                     } catch(IOException ie) {
101.                         LOG.warn("Error closing socket for handler " + this, ie);
102.                     }
103.                 }
104.                 break;
105.             }
106.         }
107.     }
108.    
109.     public void ping() {
110.         if (!sendingThreadStarted) {
111.             return;
112.         }
113.         long id;
114.         if (syncLimitCheck.check(System.nanoTime())) {
115.             synchronized(leader) {
116.                 id = leader.lastProposed;
117.             }
118.             QuorumPacket ping = new QuorumPacket(Leader.PING, id, null, null);
119.             queuePacket(ping);
120.         } else {
121.             LOG.warn("Closing connection to peer due to transaction timeout.");
122.             //Learner故障，那么就关闭当前Learner实例
123.             shutdown();
124.         }
125.     }
126.    
127.     public void shutdown() {
128.         // Send the packet of death
129.         try {
130.             queuedPackets.put(proposalOfDeath);
131.         } catch (InterruptedException e) {
132.             LOG.warn("Ignoring unexpected exception", e);
133.         }
134.         try {
135.             if (sock != null && !sock.isClosed()) {
136.                 sock.close();
137.             }
138.         } catch (IOException e) {
139.             LOG.warn("Ignoring unexpected exception during socket close", e);
140.         }
141.         this.interrupt();
142.         leader.removeLearnerHandler(this);
143.     }
144.     ...
145. }

复制代码

二.然后Leader处理Learner的事务投票响应后进行事务提交
Leader有一个HashSet为forwardingFollowers，用来管理Follower服务器。当Leader对一个事务请求发起Proposal提议的投票并发现投票通过后，也就是调用如下方法时：

1. Leader的processAck()方法 ->
2. Leader的tryToCommit()方法 ->
3. Leader的commit()方法 ->
4. Leader的sendPacket()方法

复制代码

会在Leader的sendPacket()方法中遍历forwardingFollowers里的LearnerHandler实例，将Commit请求交给Learner和Leader建立连接时生成的LearnerHandler，最后由Leader的每个LearnerHandler实例广播给对应的Learner进行事务提交。

1. //1.Leader通过Leader.propose方法对事务请求生成Proposal提议并进行广播给所有Follower
2. public class Leader {
3.     private final HashSet<LearnerHandler> forwardingFollowers = new HashSet<LearnerHandler>();
4.     final ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();
5.     ...
6.     public Proposal propose(Request request) throws XidRolloverException {
7.         ...
8.         byte[] data = SerializeUtils.serializeRequest(request);
9.         proposalStats.setLastBufferSize(data.length);
10.         QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, data, null);
11.         //生成Proposal提议
12.         Proposal p = new Proposal();
13.         p.packet = pp;
14.         p.request = request;
15.       
16.         synchronized(this) {
17.             p.addQuorumVerifier(self.getQuorumVerifier());
18.             if (request.getHdr().getType() == OpCode.reconfig) {
19.                 self.setLastSeenQuorumVerifier(request.qv, true);                       
20.             }
21.             if (self.getQuorumVerifier().getVersion()<self.getLastSeenQuorumVerifier().getVersion()) {
22.                 p.addQuorumVerifier(self.getLastSeenQuorumVerifier());
23.             }
24.             lastProposed = p.packet.getZxid();
25.             //将发送的Proposal提议放入outstandingProposals队列中
26.             outstandingProposals.put(lastProposed, p);
27.             //发送Proposal提议，其实就是把Proposal提议交给LearnerHandler处理
28.             sendPacket(pp);
29.         }
30.         return p;
31.     }
32.    
33.     void sendPacket(QuorumPacket qp) {
34.         synchronized (forwardingFollowers) {
35.             for (LearnerHandler f : forwardingFollowers) {
36.                 //LearnerHandler会将提议放入其发送队列里
37.                 f.queuePacket(qp);
38.             }
39.         }
40.     }
41.     ...
42. }
44. //2.Leader完成事务日志记录后，便会通过Leader.processAck方法记录Leader已对Proposal提议完成投票
45. //SyncRequestProcessor的nextProcessor就是AckRequestProcessor
46. class AckRequestProcessor implements RequestProcessor {
47.     ...
48.     public void processRequest(Request request) {
49.         ...
50.         //Leader也作为参与Proposal投票的一份子进行ACK响应
51.         //将Leader的SID添加到Proposal提议的投票收集器里 + 检查Proposal提议的投票收集器是否有过半ACK才提交
52.         leader.processAck(self.getId(), request.zxid, null);
53.         ...
54.     }
55. }
57. //3.Follower收到提议的投票请求后返回ACK响应给Leader
58. //Leader接收到Follower的ACK响应后，便会通过Leader.processAck方法记录该Follower已对提议完成投票
59. public class LearnerHandler extends ZooKeeperThread {
60.     //The packets to be sent to the learner
61.     final LinkedBlockingQueue<QuorumPacket> queuedPackets = new LinkedBlockingQueue<QuorumPacket>();
62.     ...
63.     @Override
64.     public void run() {
65.         ...
66.         startSendingPackets();//开启一个线程发送queuedPackets里的Packet给Learner
67.         ...
68.         while (true) {
69.             qp = new QuorumPacket();
70.             ia.readRecord(qp, "packet");//读取Learner的响应
71.             ...
72.             switch (qp.getType()) {
73.                 case Leader.ACK:
74.                     ...
75.                     //如果Leader收到Follower对某Proposal提议请求返回的ACK响应
76.                     //那么就将Follower的SID添加到该Proposal提议的投票收集器里
77.                     leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());
78.                     break;
79.                 ...
80.             }
81.         }
82.     }
83.    
84.     protected void startSendingPackets() {
85.         ...
86.         new Thread() {
87.             public void run() {
88.                 Thread.currentThread().setName("Sender-" + sock.getRemoteSocketAddress());
89.                 sendPackets();
90.             }
91.         }.start();
92.         ...
93.     }
94.    
95.     private void sendPackets() throws InterruptedException {
96.         while (true) {
97.             ...
98.             QuorumPacket p = queuedPackets.poll();
99.             oa.writeRecord(p, "packet");
100.             ...
101.         }
102.     }
103.     ...
104. }
106. public class Leader {
107.     private final HashSet<LearnerHandler> forwardingFollowers = new HashSet<LearnerHandler>();
108.     final ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();
109.     ...
110.     synchronized public void processAck(long sid, long zxid, SocketAddress followerAddr) {        
111.         ...
112.         //检查请求的ZXID，需要比上次已提交的请求的ZXID也就是lastCommitted要大
113.         if (lastCommitted >= zxid) {
114.             if (LOG.isDebugEnabled()) {
115.                 LOG.debug("proposal has already been committed, pzxid: 0x{} zxid: 0x{}", Long.toHexString(lastCommitted), Long.toHexString(zxid));
116.             }
117.             // The proposal has already been committed
118.             return;
119.         }
120.         Proposal p = outstandingProposals.get(zxid);
121.         //将Leader的SID添加到Proposal提议的投票收集器里
122.         p.addAck(sid);
123.         //尝试提交，即检查Proposal提议的投票收集器中是否有过半ACK响应
124.         boolean hasCommitted = tryToCommit(p, zxid, followerAddr);
125.         ...
126.     }
127.    
128.     synchronized public boolean tryToCommit(Proposal p, long zxid, SocketAddress followerAddr) {      
129.         //如果提议队列中存在该提议的前一个提议，说明该提议的前一个提议还没提交，那么就返回false
130.         if (outstandingProposals.containsKey(zxid - 1)) return false;
131.         //getting a quorum from all necessary configurations.
132.         //Proposal提议的投票收集器是否已过半
133.         if (!p.hasAllQuorums()) {
134.             return false;                 
135.         }
136.         ...
137.         outstandingProposals.remove(zxid);
138.         if (p.request != null) {
139.             toBeApplied.add(p);
140.         }
141.         ...
142.         //一旦提议通过，马上就要在Leader中标记lastCommitted为最新的提交ZXID
143.         commit(zxid);//给Follower广播commit消息
144.         inform(p);//给Observer发送commit消息
145.         ...
146.         //调用CommitProcessor处理器的commit方法提交请求
147.         zk.commitProcessor.commit(p.request);//让Leader执行commit消息
148.         //下面处理的是Learner发起的同步请求
149.         if (pendingSyncs.containsKey(zxid)) {
150.             for (LearnerSyncRequest r: pendingSyncs.remove(zxid)) {
151.                 sendSync(r);
152.             }               
153.         }
154.         return  true;   
155.     }
156.    
157.     //广播commit消息
158.     public void commit(long zxid) {
159.         synchronized(this) {
160.             //标记lastCommitted为最新的提交ZXID
161.             lastCommitted = zxid;
162.         }
163.         QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
164.         sendPacket(qp);
165.     }
166.    
167.     void sendPacket(QuorumPacket qp) {
168.         synchronized (forwardingFollowers) {
169.             for (LearnerHandler f : forwardingFollowers) {
170.                 //调用LearnerHandler的queuePacket方法添加Packet到发送队列
171.                 f.queuePacket(qp);
172.             }
173.         }
174.     }
175.    
176.     public void inform(Proposal proposal) {
177.         QuorumPacket qp = new QuorumPacket(Leader.INFORM, proposal.request.zxid, proposal.packet.getData(), null);
178.         sendObserverPacket(qp);
179.     }
180.     ...
181.     static public class Proposal extends SyncedLearnerTracker {
182.         public QuorumPacket packet;
183.         public Request request;
184.         ...
185.     }
186. }
188. public class SyncedLearnerTracker {
189.     protected ArrayList<QuorumVerifierAcksetPair> qvAcksetPairs = new ArrayList<QuorumVerifierAcksetPair>();
190.     ...
191.     //添加到投票收集器
192.     public boolean addAck(Long sid) {
193.         boolean change = false;
194.         for (QuorumVerifierAcksetPair qvAckset : qvAcksetPairs) {
195.             if (qvAckset.getQuorumVerifier().getVotingMembers().containsKey(sid)) {
196.                 qvAckset.getAckset().add(sid);
197.                 change = true;
198.             }
199.         }
200.         return change;
201.     }
202.    
203.     //判断投票收集器是否过半
204.     public boolean hasAllQuorums() {
205.         for (QuorumVerifierAcksetPair qvAckset : qvAcksetPairs) {
206.             if (!qvAckset.getQuorumVerifier().containsQuorum(qvAckset.getAckset()))
207.                 return false;
208.         }
209.         return true;
210.     }
211.     ...
212. }

复制代码

(3)实现数据一致性的恢复模式
当Leader故障时，Follower服务器会发生如下操作：首先Follower的followLeader()方法里的while循环会被中断运行，然后在QuorumPeer线程中就会触发执行Follower的shutdown()方法，接着执行QuorumPeer的updateServerState()方法更改节点的状态为LOOKING，之后Follower服务器在QuorumPeer线程中会重新进行Leader选举。
 
重新选举Leader需要经历一段时间，此时集群会短暂没有Leader服务器，而且重新选举Leader期间，Follower也会被关闭。
 
注意：Leader故障时，ZooKeeperServer的shutdown()方法会关闭firstProcessor线程。所以恢复模式下的选举过程中，发送到Learner的请求会进入firstProcessor，但是这些请求都会先被queuedRequests存起来，暂时不处理。

1. public class QuorumPeerMain {
2.     protected QuorumPeer quorumPeer;
3.    
4.     public void runFromConfig(QuorumPeerConfig config) throws IOException, AdminServerException {
5.         ...
6.         quorumPeer.start();
7.         quorumPeer.join();
8.         ...
9.     }
10.     ...
11. }
13. public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider {
14.     ...
15.     public synchronized void start() {
16.         loadDataBase();
17.         startServerCnxnFactory();
18.         adminServer.start();
19.         //初始化Leader选举(初始化当前投票+监听选举端口+启动选举守护线程)
20.         startLeaderElection();
21.         startJvmPauseMonitor();
22.         super.start();
23.     }
24.    
25.     @Override
26.     public void run() {
27.         ...
28.         while (running) {
29.             switch (getPeerState()) {
30.                 case LOOKING:
31.                     ...
32.                     if (shuttingDownLE) {
33.                         shuttingDownLE = false;
34.                         startLeaderElection();
35.                     }
36.                     //调用QuorumPeer.electionAlg的lookForLeader()，也就是FastLeaderElection.lookForLeader()开启一轮选举
37.                     setCurrentVote(makeLEStrategy().lookForLeader());
38.                     ...
39.                 case FOLLOWING:
40.                     try {
41.                         LOG.info("FOLLOWING");
42.                         setFollower(makeFollower(logFactory));
43.                         //Leader故障，那么就会中断follower.followLeader()里的的while循环
44.                         follower.followLeader();
45.                     } catch (Exception e) {
46.                         LOG.warn("Unexpected exception",e);
47.                     } finally {
48.                         //Leader故障，就会执行这里的方法
49.                         follower.shutdown();
50.                         setFollower(null);
51.                         updateServerState();
52.                     }
53.                     break;
54.                 case LEADING:
55.                     LOG.info("LEADING");
56.                     try {
57.                         setLeader(makeLeader(logFactory));
58.                         leader.lead();
59.                         setLeader(null);
60.                     } catch (Exception e) {
61.                         LOG.warn("Unexpected exception",e);
62.                     } finally {
63.                         if (leader != null) {
64.                             leader.shutdown("Forcing shutdown");
65.                             setLeader(null);
66.                         }
67.                         updateServerState();
68.                     }
69.                     break;
70.             }
71.             ...
72.         }
73.         ...
74.     }
75.    
76.     private synchronized void updateServerState() {
77.         //reconfigFlag初始化就为false
78.         if (!reconfigFlag) {
79.             setPeerState(ServerState.LOOKING);
80.             LOG.warn("PeerState set to LOOKING");
81.             return;
82.         }
83.         ...
84.         //updateServerStatez方法被执行后又会重置为false
85.         reconfigFlag = false;
86.     }
87.    
88.     //开始Leader选举
89.     //创建选举服务端QuorumCnxManager并启动监听 + 创建选举算法FastLeaderElection并启动
90.     //将FastLeaderElection实例赋值给QuorumPeer.electionAlg
91.     synchronized public void startLeaderElection() {
92.         if (getPeerState() == ServerState.LOOKING) {
93.             currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
94.         }
95.         ...
96.         this.electionAlg = createElectionAlgorithm(electionType);
97.     }
98.     ...
99. }
101. public class Follower extends Learner{
102.     ...
103.     public void shutdown() {   
104.         super.shutdown();
105.     }
106.    
107.     void followLeader() throws InterruptedException {
108.         ...
109.         QuorumServer leaderServer = findLeader();
110.         try {
111.             //Follower启动时向Leader发起连接
112.             connectToLeader(leaderServer.addr, leaderServer.hostname);
113.             long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);
114.             ...
115.             syncWithLeader(newEpochZxid);
116.             ...
117.             QuorumPacket qp = new QuorumPacket();
118.             while (this.isRunning()) {
119.                 //读取BIO输入流
120.                 readPacket(qp);
121.                 processPacket(qp);
122.             }  
123.         } catch (Exception e) {
124.             //Leader故障，那么这里就会报异常，从而中断上面的while循环
125.             LOG.warn("Exception when following the leader", e);
126.             closeSocket();
127.             pendingRevalidations.clear();
128.         }
129.         ...
130.     }
131.     ...
132. }
134. public class Learner {      
135.     ...
136.     public void shutdown() {
137.         self.setZooKeeperServer(null);
138.         self.closeAllConnections();
139.         self.adminServer.setZooKeeperServer(null);
140.         closeSocket();
141.         //关闭比如FollowerZooKeeperServer
142.         if (zk != null) {
143.             //关闭sessionTracker的超时检查线程 + 设置firstProcessor为null + 清空zkDb
144.             zk.shutdown(self.getSyncMode().equals(QuorumPeer.SyncMode.SNAP));
145.         }
146.     }
147.    
148.     protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception {
149.         //创建Learner当前BIO的客户端Socket
150.         this.sock = createSocket();
151.         int initLimitTime = self.tickTime * self.initLimit;
152.         int remainingInitLimitTime = initLimitTime;
153.         long startNanoTime = nanoTime();
154.         //尝试重连最多5次
155.         for (int tries = 0; tries < 5; tries++) {
156.             try {
157.                 remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000);
158.                 if (remainingInitLimitTime <= 0) {
159.                     LOG.error("initLimit exceeded on retries.");
160.                     throw new IOException("initLimit exceeded on retries.");
161.                 }
162.                 //向Leader发起连接请求
163.                 sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime));
164.                 if (self.isSslQuorum())  {
165.                     ((SSLSocket) sock).startHandshake();
166.                 }
167.                 sock.setTcpNoDelay(nodelay);
168.                 break;
169.             } catch (IOException e) {
170.                 ...
171.             }
172.             Thread.sleep(1000);
173.         }
174.         self.authLearner.authenticate(sock, hostname);
175.         //初始化输入输出流
176.         leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream()));
177.         bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
178.         leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);
179.     }
180.    
181.     private Socket createSocket() throws X509Exception, IOException {
182.         //创建客户端Socket
183.         Socket sock;
184.         if (self.isSslQuorum()) {
185.             sock = self.getX509Util().createSSLSocket();
186.         } else {
187.             sock = new Socket();
188.         }
189.         sock.setSoTimeout(self.tickTime * self.initLimit);
190.         return sock;
191.     }
192.    
193.     protected void sockConnect(Socket sock, InetSocketAddress addr, int timeout) throws IOException {
194.         //向BIO的服务端发起连接请求
195.         sock.connect(addr, timeout);
196.     }
197.    
198.     void readPacket(QuorumPacket pp) throws IOException {
199.         synchronized (leaderIs) {
200.             leaderIs.readRecord(pp, "packet");
201.         }
202.     }
203.     ...
204. }
206. public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
207.     ...
208.     public synchronized void shutdown(boolean fullyShutDown) {
209.         ...
210.         //关闭会话的超时检查线程
211.         if (sessionTracker != null) {
212.             sessionTracker.shutdown();
213.         }
214.         //关闭firstProcessor线程
215.         //所以Leader故障时，发送到Learner的请求会在firstProcessor的queuedRequests中存起来，暂时不处理
216.         if (firstProcessor != null) {
217.             firstProcessor.shutdown();
218.         }
219.         if (zkDb != null) {
220.             if (fullyShutDown) {
221.                 zkDb.clear();
222.             } else {
223.                 ...
224.             }
225.         }
226.         unregisterJMX();
227.     }
228.     ...
229. }

复制代码

 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！