zookeeper源码(04)leader选举流程

在"zookeeper源码(03)集群启动流程"中介绍了leader选举的入口，本文将详细分析leader选举组件和流程。
leader选举流程(重要)

• quorumPeer的start阶段使用startLeaderElection()方法启动选举
  
• LOOKING状态，投自己一票
  
• createElectionAlgorithm - 创建选举核心组件：QuorumCnxManager(管理连接)、FastLeaderElection(选举)等
  
• quorumPeer的main loop根据当前状态执行不同流程
  

状态与流程：

• LOOKING - 使用fastLeaderElection.lookForLeader选举
  
  
  
  • 递增选举epoch开启新一轮选举
    
  • 使用自己的serverId、zxid、currentEpoch初始化投票决议
    
  • 把选票发出去
    
  • 循环接收其他server的选票：
    
    
    
    • LOOKING选票：对比选举epoch、currentEpoch、zxid、serverId决定投给哪个server，若是超过半数节点同意该决议，则将该server确定为leader
      
    • FOLLOWING选票：对比选举epoch后将选票投给当前leader
      
    • LEADING选票：对比选举epoch后将选票投给当前leader
      
    
    
  
  
• LEADING - 创建Leader对象执行lead逻辑
  
  
  
  • zkServer加载数据
    
  • 启动quorum监听
    
  • 根据各个follower的当前epoch确定新的epoch和zxid
    
  • 给follower同步数据
    
  • 启动zkServer
    
  • 每间隔tick验证多数follower同步状态
    
  
  
• FOLLOWING - 创建Follower对象指定followLeader逻辑
  
  
  
  • connectToLeader - 连接leader服务器
    
  • registerWithLeader - 向leader发送当前epoch，等待leader发送新一轮的epoch
    
  • syncWithLeader - 接收leader同步的数据：txnlog、committedlog、snapshot
    
  • 保持通信处理来自leader的数据包
    
  
  
• OBSERVING - 创建Observer对象执行observeLeader逻辑，基本与FOLLOWING相同
  

启动leader选举

QuorumPeer的startLeaderElection方法是启动选举的入口：

1. public synchronized void startLeaderElection() {
2.     try {
3.         if (getPeerState() == ServerState.LOOKING) {
4.             // 投自己一票，封装zxid和epoch
5.             currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
6.         }
7.     } catch (IOException e) {
8.         RuntimeException re = new RuntimeException(e.getMessage());
9.         re.setStackTrace(e.getStackTrace());
10.         throw re;
11.     }
12.     // electionType总是3
13.     this.electionAlg = createElectionAlgorithm(electionType);
14. }
16. protected Election createElectionAlgorithm(int electionAlgorithm) {
17.     Election le = null;
19.     // TODO: use a factory rather than a switch
20.     // 可以使用策略模式替换switch语句
21.     switch (electionAlgorithm) {
22.     case 1:
23.         throw new UnsupportedOperationException("Election Algorithm 1 is not supported.");
24.     case 2:
25.         throw new UnsupportedOperationException("Election Algorithm 2 is not supported.");
26.     case 3:
27.         QuorumCnxManager qcm = createCnxnManager();
28.         QuorumCnxManager oldQcm = qcmRef.getAndSet(qcm);
29.         // 关闭oldQcm
30.         if (oldQcm != null) {
31.             oldQcm.halt();
32.         }
33.         // 用来启动serverSocket监听
34.         QuorumCnxManager.Listener listener = qcm.listener;
35.         if (listener != null) {
36.             listener.start();
37.             FastLeaderElection fle = new FastLeaderElection(this, qcm);
38.             fle.start();
39.             le = fle;
40.         }
41.         break;
42.     default:
43.         assert false;
44.     }
45.     return le;
46. }
48. public QuorumCnxManager createCnxnManager() {
49.     // socket超时设置使用，默认tickTime * syncLimit
50.     // 按照zoo_sample.cfg文件配置是2000 * 5
51.     int timeout = quorumCnxnTimeoutMs > 0 ? quorumCnxnTimeoutMs : this.tickTime * this.syncLimit;
52.     return new QuorumCnxManager(
53.         this,
54.         this.getMyId(),
55.         this.getView(), // serverId->quorumServer
56.         this.authServer,
57.         this.authLearner,
58.         timeout,
59.         this.getQuorumListenOnAllIPs(), // 是否监听所有IP默认false
60.         this.quorumCnxnThreadsSize, // 默认20
61.         this.isQuorumSaslAuthEnabled());
62. }

复制代码

QuorumCnxManager类

概述：

1. This class implements a connection manager for leader election using TCP.
2. It maintains one connection for every pair of servers. The tricky part is to guarantee that there is exactly one connection for every pair of servers that are operating correctly and that can communicate over the network. If two servers try to start a connection concurrently, then the connection manager uses a very simple tie-breaking mechanism to decide which connection to drop based on the IP addressed of the two parties.
3. For every peer, the manager maintains a queue of messages to send. If the connection to any particular peer drops, then the sender thread puts the message back on the list. As this implementation currently uses a queue implementation to maintain messages to send to another peer, we add the message to the tail of the queue, thus changing the order of messages. Although this is not a problem for the leader election, it could be a problem when consolidating peer communication. This is to be verified, though.

复制代码

• 维护leader选举时server之间的tcp连接
  
• 确保两个server之间存在一个连接，如果两个server同时建立连接，则始终保留id大的一方建立的连接
  
• 使用队列缓存待发送的消息
  

主要字段

1. // 用于执行QuorumConnectionReqThread和QuorumConnectionReceiverThread
2. private ThreadPoolExecutor connectionExecutor;
4. // 管理sid -> SendWorker/BlockingQueue/ByteBuffer
5. final ConcurrentHashMap<Long, SendWorker> senderWorkerMap;
6. final ConcurrentHashMap<Long, BlockingQueue<ByteBuffer>> queueSendMap;
7. final ConcurrentHashMap<Long, ByteBuffer> lastMessageSent;
9. // 接收队列
10. public final BlockingQueue<Message> recvQueue;

复制代码

主要方法

1. public void initiateConnection(final MultipleAddresses electionAddr, final Long sid);
2. // 将initiateConnection方法放到了QuorumConnectionReqThread中然后提交给connectionExecutor异步执行
3. public boolean initiateConnectionAsync(final MultipleAddresses electionAddr, final Long sid);
5. private boolean startConnection(Socket sock, Long sid) throws IOException;
7. public void receiveConnection(final Socket sock);
8. // 将receiveConnection方法放到了QuorumConnectionReceiverThread中然后提交给connectionExecutor异步执行
9. public void receiveConnectionAsync(final Socket sock);
11. public void toSend(Long sid, ByteBuffer b);
13. boolean connectOne(long sid, MultipleAddresses electionAddr);
14. void connectOne(long sid);
15. public void connectAll();

复制代码

其余工具方法不分析。
initiateConnection方法

创建Socket对象，如有必要则做ssl握手和认证，发送初始化数据包。如果自己id小则关闭连接，以确保两个server之间存在一个连接。

1. public void initiateConnection(final MultipleAddresses electionAddr, final Long sid) {
2.     Socket sock = null;
3.     try {
4.         // 创建Socket
5.         if (self.isSslQuorum()) {
6.             sock = self.getX509Util().createSSLSocket();
7.         } else {
8.             sock = SOCKET_FACTORY.get();
9.         }
10.         setSockOpts(sock); // socket设置例如timeout
11.         // 连接目标peer
12.         sock.connect(electionAddr.getReachableOrOne(), cnxTO);
13.         // ssl握手
14.         if (sock instanceof SSLSocket) {
15.             SSLSocket sslSock = (SSLSocket) sock;
16.             sslSock.startHandshake();
17.         }
18.     } catch (X509Exception e) {
19.         closeSocket(sock);
20.         return;
21.     } catch (UnresolvedAddressException | IOException e) {
22.         closeSocket(sock);
23.         return;
24.     }
26.     try {
27.         // 发连接初始化数据包、sasl认证
28.         // 如果selfId小于对方，关闭连接
29.         // 创建SendWorker、RecvWorker并启动
30.         // 创建对应sid的发送队列
31.         startConnection(sock, sid);
32.     } catch (IOException e) {
33.         closeSocket(sock);
34.     }
35. }

复制代码

startConnection方法

• 发连接初始化数据包、sasl认证
  
• 如果selfId小于对方，关闭连接
  
• 创建SendWorker、RecvWorker并启动
  
• 创建对应sid的发送队列
  

1. private boolean startConnection(Socket sock, Long sid) throws IOException {
2.     DataOutputStream dout = null;
3.     DataInputStream din = null;
4.     try {
5.         // 输出流
6.         BufferedOutputStream buf = new BufferedOutputStream(sock.getOutputStream());
7.         dout = new DataOutputStream(buf);
9.         // 发协议版本、myid、address初始化数据包
10.         long protocolVersion = self.isMultiAddressEnabled() ? PROTOCOL_VERSION_V2 : PROTOCOL_VERSION_V1;
11.         dout.writeLong(protocolVersion);
12.         dout.writeLong(self.getMyId());
14.         // now we send our election address. For the new protocol version, we can send multiple addresses.
15.         Collection<InetSocketAddress> addressesToSend = protocolVersion == PROTOCOL_VERSION_V2
16.                 ? self.getElectionAddress().getAllAddresses()
17.                 : Arrays.asList(self.getElectionAddress().getOne());
19.         String addr = addressesToSend.stream()
20.                 .map(NetUtils::formatInetAddr).collect(Collectors.joining("|"));
21.         byte[] addr_bytes = addr.getBytes();
22.         dout.writeInt(addr_bytes.length);
23.         dout.write(addr_bytes);
24.         dout.flush();
26.         din = new DataInputStream(new BufferedInputStream(sock.getInputStream()));
27.     } catch (IOException e) {
28.         closeSocket(sock);
29.         return false;
30.     }
32.     // authenticate learner
33.     QuorumPeer.QuorumServer qps = self.getVotingView().get(sid);
34.     if (qps != null) {
35.         authLearner.authenticate(sock, qps.hostname);
36.     }
38.     if (sid > self.getMyId()) { // If lost the challenge, then drop the new connection
39.         closeSocket(sock);
40.     } else {
41.         // 创建SendWorker、RecvWorker
42.         SendWorker sw = new SendWorker(sock, sid);
43.         RecvWorker rw = new RecvWorker(sock, din, sid, sw);
44.         sw.setRecv(rw);
46.         SendWorker vsw = senderWorkerMap.get(sid);
48.         if (vsw != null) {
49.             vsw.finish();
50.         }
52.         senderWorkerMap.put(sid, sw);
54.         // 创建发送队列
55.         queueSendMap.putIfAbsent(sid, new CircularBlockingQueue<>(SEND_CAPACITY));
57.         sw.start();
58.         rw.start();
60.         return true;
61.     }
62.     return false;
63. }

复制代码

receiveConnection方法

当server收到连接请求，如果change获胜(selfId大于对方)，将关闭该连接，由自己去连接对方。

1. public void receiveConnection(final Socket sock) {
2.     DataInputStream din = null;
3.     try {
4.         // 输入流
5.         din = new DataInputStream(new BufferedInputStream(sock.getInputStream()));
6.         handleConnection(sock, din);
7.     } catch (IOException e) {
8.         closeSocket(sock);
9.     }
10. }
12. private void handleConnection(Socket sock, DataInputStream din) throws IOException {
13.     Long sid = null, protocolVersion = null;
14.     MultipleAddresses electionAddr = null;
16.     try {
17.         protocolVersion = din.readLong();
18.         if (protocolVersion >= 0) { // this is a server id and not a protocol version
19.             sid = protocolVersion;
20.         } else {
21.             try {
22.                 InitialMessage init = InitialMessage.parse(protocolVersion, din);
23.                 sid = init.sid;
24.                 if (!init.electionAddr.isEmpty()) {
25.                     electionAddr = new MultipleAddresses(init.electionAddr,
26.                             Duration.ofMillis(self.getMultiAddressReachabilityCheckTimeoutMs()));
27.                 }
28.             } catch (InitialMessage.InitialMessageException ex) {
29.                 closeSocket(sock);
30.                 return;
31.             }
32.         }
34.         if (sid == QuorumPeer.OBSERVER_ID) {
35.             // Choose identifier at random. We need a value to identify the connection.
36.             sid = observerCounter.getAndDecrement();
37.         }
38.     } catch (IOException e) {
39.         closeSocket(sock);
40.         return;
41.     }
43.     // do authenticating learner
44.     authServer.authenticate(sock, din);
45.     // If wins the challenge, then close the new connection.
46.     if (sid < self.getMyId()) { // 对方比自己id小，需要关闭当前连接，由自己去连接对方
47.         SendWorker sw = senderWorkerMap.get(sid);
48.         if (sw != null) {
49.             sw.finish();
50.         }
52.         // 关闭连接
53.         closeSocket(sock);
55.         if (electionAddr != null) {
56.             connectOne(sid, electionAddr); // 连接对方
57.         } else {
58.             connectOne(sid);
59.         }
60.     } else if (sid == self.getMyId()) {
61.     } else { // 创建SendWorker、RecvWorker和发送队列
62.         SendWorker sw = new SendWorker(sock, sid);
63.         RecvWorker rw = new RecvWorker(sock, din, sid, sw);
64.         sw.setRecv(rw);
66.         SendWorker vsw = senderWorkerMap.get(sid);
68.         if (vsw != null) {
69.             vsw.finish();
70.         }
72.         senderWorkerMap.put(sid, sw);
74.         queueSendMap.putIfAbsent(sid, new CircularBlockingQueue<>(SEND_CAPACITY));
76.         sw.start();
77.         rw.start();
78.     }
79. }

复制代码

toSend方法

发消息。

1. public void toSend(Long sid, ByteBuffer b) {
2.     // 如果是给自己的消息，直接发给recvQueue
3.     if (this.mySid == sid) {
4.         b.position(0);
5.         addToRecvQueue(new Message(b.duplicate(), sid));
6.     } else {
7.         // 将消息发给sid对应的发送队列
8.         BlockingQueue<ByteBuffer> bq =
9.             queueSendMap.computeIfAbsent(sid, serverId -> new CircularBlockingQueue<>(SEND_CAPACITY));
10.         addToSendQueue(bq, b);
11.         // 检查是否建立了连接
12.         connectOne(sid);
13.     }
14. }

复制代码

connectOne方法

1. synchronized boolean connectOne(long sid, MultipleAddresses electionAddr) {
2.     // 已经建立过连接
3.     if (senderWorkerMap.get(sid) != null) {
4.         if (self.isMultiAddressEnabled() && electionAddr.size() > 1 &&
5.             self.isMultiAddressReachabilityCheckEnabled()) {
6.             // check是否可达
7.             senderWorkerMap.get(sid).asyncValidateIfSocketIsStillReachable();
8.         }
9.         return true;
10.     }
11.     // 异步建立新连接
12.     return initiateConnectionAsync(electionAddr, sid);
13. }
15. synchronized void connectOne(long sid) {
16.     if (senderWorkerMap.get(sid) != null) {
17.         if (self.isMultiAddressEnabled() && self.isMultiAddressReachabilityCheckEnabled()) {
18.             senderWorkerMap.get(sid).asyncValidateIfSocketIsStillReachable();
19.         }
20.         return;
21.     }
22.     // 使用sid从lastCommittedView、lastProposedView中解析address之后在建立连接
23.     synchronized (self.QV_LOCK) {
24.         boolean knownId = false;
25.         // Resolve hostname for the remote server before attempting to
26.         // connect in case the underlying ip address has changed.
27.         self.recreateSocketAddresses(sid);
28.         Map<Long, QuorumPeer.QuorumServer> lastCommittedView = self.getView();
29.         QuorumVerifier lastSeenQV = self.getLastSeenQuorumVerifier();
30.         Map<Long, QuorumPeer.QuorumServer> lastProposedView = lastSeenQV.getAllMembers();
31.         if (lastCommittedView.containsKey(sid)) {
32.             knownId = true;
33.             if (connectOne(sid, lastCommittedView.get(sid).electionAddr)) {
34.                 return;
35.             }
36.         }
37.         if (lastSeenQV != null
38.             && lastProposedView.containsKey(sid)
39.             && (!knownId ||
40.                 !lastProposedView.get(sid).electionAddr.equals(lastCommittedView.get(sid).electionAddr))) {
41.             knownId = true;
42.             if (connectOne(sid, lastProposedView.get(sid).electionAddr)) {
43.                 return;
44.             }
45.         }
46.     }
47. }

复制代码

connectAll方法

Try to establish a connection with each server if one doesn't exist.

1. public void connectAll() {
2.     long sid;
3.     for (Enumeration<Long> en = queueSendMap.keys(); en.hasMoreElements(); ) {
4.         sid = en.nextElement();
5.         connectOne(sid);
6.     }
7. }

复制代码

Listener类

用来启动serverSocket监听，一个线程类，在run方法启动监听：

1. public void run() {
2.     if (!shutdown) {
3.         Set<InetSocketAddress> addresses;
5.         // 获取需要监听的地址
6.         if (self.getQuorumListenOnAllIPs()) {
7.             addresses = self.getElectionAddress().getWildcardAddresses();
8.         } else {
9.             addresses = self.getElectionAddress().getAllAddresses();
10.         }
11.         // 用于阻塞等待
12.         CountDownLatch latch = new CountDownLatch(addresses.size());
13.         // 为每一个监听地址创建ListenerHandler
14.         listenerHandlers = addresses.stream().map(address ->
15.                         new ListenerHandler(address,self.shouldUsePortUnification(),
16.                                             self.isSslQuorum(), latch))
17.                 .collect(Collectors.toList());
19.         final ExecutorService executor = Executors.newFixedThreadPool(addresses.size());
20.         try {
21.             // 启动ListenerHandler
22.             listenerHandlers.forEach(executor::submit);
23.         } finally {
24.             executor.shutdown();
25.         }
27.         try {
28.             // 阻塞等待，ListenerHandler结束之后会countdown
29.             latch.await();
30.         } catch (InterruptedException ie) {
31.         } finally {
32.             // Clean up for shutdown 略
33.         }
34.     }
35.     // 略
36. }

复制代码

ListenerHandler run方法：

1. public void run() {
2.     try {
3.         // 接受连接
4.         acceptConnections();
5.         try {
6.             close();
7.         } catch (IOException e) {}
8.     } catch (Exception e) {
9.     } finally {
10.         latch.countDown();
11.     }
12. }
14. private void acceptConnections() {
15.     int numRetries = 0;
16.     Socket client = null;
18.     while ((!shutdown) && (portBindMaxRetry == 0 || numRetries < portBindMaxRetry)) {
19.         try {
20.             // 创建ServerSocket并bind端口
21.             serverSocket = createNewServerSocket();
22.             while (!shutdown) {
23.                 try {
24.                     // 接受客户端Socket
25.                     client = serverSocket.accept();
26.                     setSockOpts(client); // socket设置如timeout
27.                     // 使用receiveConnection处理新的连接
28.                     if (quorumSaslAuthEnabled) {
29.                         receiveConnectionAsync(client);
30.                     } else {
31.                         receiveConnection(client);
32.                     }
33.                     numRetries = 0;
34.                 } catch (SocketTimeoutException e) {}
35.             }
36.         } catch (IOException e) {
37.             // 略
38.         }
39.     }
40.     // 略
41. }

复制代码

QuorumConnectionReqThread类

用于异步连接其他peer服务，run方法调用initiateConnection方法建立连接。
QuorumConnectionReceiverThread类

用于异步接受连接，run方法调用receiveConnection方法处理新建立的连接。
SendWorker类

Thread to send messages. Instance waits on a queue, and send a message as soon as there is one available. If connection breaks, then opens a new one.
用来发送消息的线程：

• 封装sid、socket、连接输出流
  
• 从发送队列取消息，通过输出流发送
  

RecvWorker类

Thread to receive messages. Instance waits on a socket read. If the channel breaks, then removes itself from the pool of receivers.
用来读取消息的线程：

1. public void run() {
2.     threadCnt.incrementAndGet();
3.     try {
4.         while (running && !shutdown && sock != null) {
5.             // 读取消息长度
6.             int length = din.readInt();
7.             if (length <= 0 || length > PACKETMAXSIZE) {
8.                 throw new IOException("Received packet with invalid packet: " + length);
9.             }
10.             // 读取数据
11.             final byte[] msgArray = new byte[length];
12.             din.readFully(msgArray, 0, length);
13.             // 保存到接收队列
14.             addToRecvQueue(new Message(ByteBuffer.wrap(msgArray), sid));
15.         }
16.     } catch (Exception e) {
17.     } finally {
18.         sw.finish();
19.         closeSocket(sock);
20.     }
21. }

复制代码

FastLeaderElection类

文档说明：

1. Implementation of leader election using TCP. It uses an object of the class QuorumCnxManager to manage connections. Otherwise, the algorithm is push-based as with the other UDP implementations. There are a few parameters that can be tuned to change its behavior. First, finalizeWait determines the amount of time to wait until deciding upon a leader. This is part of the leader election algorithm.

复制代码

• 使用tcp实现leader选举，基于推送模式
  
• 使用QuorumCnxManager对象管理连接
  

构造方法

1. public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager) {
2.     this.stop = false;
3.     this.manager = manager;
4.     starter(self, manager);
5. }
7. private void starter(QuorumPeer self, QuorumCnxManager manager) {
8.     this.self = self;
9.     proposedLeader = -1;
10.     proposedZxid = -1;
12.     sendqueue = new LinkedBlockingQueue<>();
13.     recvqueue = new LinkedBlockingQueue<>();
14.     // 用来启动WorkerSender和WorkerReceiver
15.     this.messenger = new Messenger(manager);
16. }

复制代码

主要字段

1. // 在leader最终确定之前尝试拉取变化选票的时长
2. static final int finalizeWait = 200;
4. // 投票箱，用于保存一轮选举的结果、统计选举结果
5. private SyncedLearnerTracker leadingVoteSet;
7. // 发送队列
8. LinkedBlockingQueue<ToSend> sendqueue;
9. // 接收队列
10. LinkedBlockingQueue<Notification> recvqueue;
12. // 用来启动WorkerSender和WorkerReceiver
13. Messenger messenger;
15. // 决议leaderId
16. long proposedLeader;
17. // 决议zxid
18. long proposedZxid;
19. // 决议epoch
20. long proposedEpoch;

复制代码

start方法启动选举

1. public void start() {
2.     this.messenger.start(); // 会启动WorkerSender和WorkerReceiver两个线程
3. }

复制代码

Messenger类

WorkerSender线程

• 从sendqueue取ToSend消息
  
• 通过QuorumCnxManager的toSend方法发送消息
  

WorkerReceiver线程

• 通过QuorumCnxManager的pollRecvQueue取接收的消息
  
• 封装Notification对象，推送到recvqueue队列
  

主要方法

1. // 创建发送消息
2. static ByteBuffer buildMsg(
3.     int state, long leader, long zxid, long electionEpoch, long epoch, byte[] configData);
5. // 给所有节点发Notification投票
6. private void sendNotifications();
8. // 对比serverId、zxid、currentEpoch决定将票投给哪个server
9. protected boolean totalOrderPredicate(
10.     long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch);
12. // 给定一个Vote集，返回SyncedLearnerTracker对象，用来确定是否有足够的选票确定选举结束
13. protected SyncedLearnerTracker getVoteTracker(Map<Long, Vote> votes, Vote vote);
15. // 如果有leader当选，并且有足够的选票，必须检查该leader是否投票并确认其处于领先地位
16. // 需要进行这种检查，以避免peers一次又一次地选举一个已经崩溃且不再领先的peer
17. protected boolean checkLeader(Map<Long, Vote> votes, long leader, long electionEpoch);
19. // 更新proposedLeader、proposedZxid、proposedEpoch
20. // 确定leader或者为下一轮投票做准备
21. synchronized void updateProposal(long leader, long zxid, long epoch);
23. // 使用当前proposedLeader、proposedZxid、proposedEpoch创建Vote(选票)
24. public synchronized Vote getVote();
26. // 通过zkDb获取lastLoggedZxid
27. private long getInitLastLoggedZxid();
29. // 获取currentEpoch
30. private long getPeerEpoch();
32. // 根据参数proposedLeader更新peer状态
33. // 如果已经是leader会使用voteSet更新leadingVoteSet
34. private void setPeerState(long proposedLeader, SyncedLearnerTracker voteSet);
36. // 启动一轮leader选举
37. // 当状态变为LOOKING该方法就会被调用，会给其他peer发投票notification
38. public Vote lookForLeader() throws InterruptedException;
40. // 收到FOLLOWING状态notification
41. private Vote receivedFollowingNotification(
42.     Map<Long, Vote> recvset, Map<Long, Vote> outofelection,
43.     SyncedLearnerTracker voteSet, Notification n);
45. // 收到LEADING状态notification
46. private Vote receivedLeadingNotification(
47.     Map<Long, Vote> recvset, Map<Long, Vote> outofelection,
48.     SyncedLearnerTracker voteSet, Notification n);

复制代码

buildMsg方法

1. static ByteBuffer buildMsg(int state, long leader, long zxid,
2.                            long electionEpoch, long epoch, byte[] configData) {
3.     byte[] requestBytes = new byte[44 + configData.length];
4.     ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
6.     requestBuffer.clear();
7.     requestBuffer.putInt(state); // 当前状态
8.     requestBuffer.putLong(leader); // 投票的leaderId
9.     requestBuffer.putLong(zxid); // zxid
10.     requestBuffer.putLong(electionEpoch); // 选举epoch
11.     requestBuffer.putLong(epoch); // 数据epoch
12.     requestBuffer.putInt(Notification.CURRENTVERSION); // 0x2
13.     requestBuffer.putInt(configData.length); // 数据长度
14.     requestBuffer.put(configData); // quorumVerifier数据
16.     return requestBuffer;
17. }

复制代码

totalOrderPredicate方法

对比serverId、zxid、currentEpoch决定将票投给哪个server：

1. protected boolean totalOrderPredicate(
2.     long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
4.     if (self.getQuorumVerifier().getWeight(newId) == 0) {
5.         return false;
6.     }
8.     /*
9.      * Return true if one of the following three cases hold:
10.      * 1- New epoch is higher
11.      * 2- New epoch is the same as current epoch, but new zxid is higher
12.      * 3- New epoch is the same as current epoch, new zxid is the same
13.      *  as current zxid, but server id is higher.
14.      */
15.     return ((newEpoch > curEpoch)
16.             || ((newEpoch == curEpoch)
17.                 && ((newZxid > curZxid)
18.                     || ((newZxid == curZxid)
19.                         && (newId > curId)))));
20. }

复制代码

getVoteTracker方法

给定一个Vote集，返回SyncedLearnerTracker对象，用来确定是否有足够的选票宣布选举结束：

1. protected SyncedLearnerTracker getVoteTracker(Map<Long, Vote> votes, Vote vote) {
2.     SyncedLearnerTracker voteSet = new SyncedLearnerTracker();
3.     voteSet.addQuorumVerifier(self.getQuorumVerifier());
4.     if (self.getLastSeenQuorumVerifier() != null
5.         && self.getLastSeenQuorumVerifier().getVersion() > self.getQuorumVerifier().getVersion()) {
6.         voteSet.addQuorumVerifier(self.getLastSeenQuorumVerifier());
7.     }
9.     // 比对其他server响应的选票和本地的选票，决定是否将选票sid放入ack集
10.     for (Map.Entry<Long, Vote> entry : votes.entrySet()) {
11.         if (vote.equals(entry.getValue())) {
12.             voteSet.addAck(entry.getKey()); // key是sid
13.         }
14.     }
16.     return voteSet;
17. }

复制代码

checkLeader方法

1. protected boolean checkLeader(Map<Long, Vote> votes, long leader, long electionEpoch) {
3.     boolean predicate = true;
5.     if (leader != self.getMyId()) {
6.         if (votes.get(leader) == null) { // leader服务器必须投票，否则次轮投票也无效
7.             predicate = false;
8.         } else if (votes.get(leader).getState() != ServerState.LEADING) {
9.             // leader服务器的状态必须是LEADING，否则次轮投票也无效
10.             predicate = false;
11.         }
12.     } else if (logicalclock.get() != electionEpoch) { // 选举epoch必须一致
13.         predicate = false;
14.     }
16.     return predicate;
17. }

复制代码

lookForLeader方法

启动一轮leader选举，当状态变为LOOKING该方法就会被调用，会给其他peer发投票notification通知：

1. public Vote lookForLeader() throws InterruptedException {
2.     // 略
3.     try {
4.         // 存储当前选举周期的sid -> vote选票数据
5.         Map<Long, Vote> recvset = new HashMap<>();
7.         // 存储之前选举周期的sid -> vote选票数据
8.         Map<Long, Vote> outofelection = new HashMap<>();
10.         int notTimeout = minNotificationInterval;
12.         synchronized (this) {
13.             logicalclock.incrementAndGet(); // 递增选举epoch开始新一轮选举
14.             // 初始化选举"决议"，最开始都是投票给自己
15.             updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
16.         }
17.         // 给所有节点发通知
18.         sendNotifications();
19.         // 投票箱
20.         SyncedLearnerTracker voteSet = null;
22.         // 正常情况下直到选出leader才会退出
23.         while ((self.getPeerState() == ServerState.LOOKING) && (!stop)) {
24.             Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS);
26.             if (n == null) {
27.                 // 重发或者重连
28.                 if (manager.haveDelivered()) {
29.                     sendNotifications();
30.                 } else {
31.                     manager.connectAll();
32.                 }
34.                 notTimeout = Math.min(notTimeout << 1, maxNotificationInterval);
36.                 // 略
38.             } else if (validVoter(n.sid) && validVoter(n.leader)) {
39.                 switch (n.state) {
40.                 case LOOKING:
41.                     // 略
42.                     // 对方的选举epoch比自己大
43.                     if (n.electionEpoch > logicalclock.get()) {
44.                         logicalclock.set(n.electionEpoch); // 同步为新的epoch
45.                         recvset.clear(); // 清空投票集
46.                         // 比对选票，如果对方赢了，则使用对方的选票更新到本地
47.                         if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
48.                                                 getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
49.                             updateProposal(n.leader, n.zxid, n.peerEpoch);
50.                         } else {
51.                             updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
52.                         }
53.                         // 把最新的选票发出去
54.                         sendNotifications();
55.                     } else if (n.electionEpoch < logicalclock.get()) {
56.                         // 对方的选举epoch比自己小
57.                         break;
58.                     } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
59.                                                    proposedLeader, proposedZxid, proposedEpoch)) {
60.                         updateProposal(n.leader, n.zxid, n.peerEpoch);
61.                         sendNotifications();
62.                     }
64.                     // 保存到选票集
65.                     recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
66.                     // 创建投票箱
67.                     voteSet = getVoteTracker(
68.                         recvset, new Vote(proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch));
69.                     // 判断acks>half表示已经选举出了leader
70.                     if (voteSet.hasAllQuorums()) {
72.                         // 等待拉取变化的选票
73.                         while ((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null) {
74.                             if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
75.                                                     proposedLeader, proposedZxid, proposedEpoch)) {
76.                                 recvqueue.put(n);
77.                                 break;
78.                             }
79.                         }
81.                         // 设置peer状态
82.                         if (n == null) {
83.                             setPeerState(proposedLeader, voteSet);
84.                             Vote endVote = new Vote(
85.                                 proposedLeader, proposedZxid, logicalclock.get(), proposedEpoch);
86.                             leaveInstance(endVote);
87.                             return endVote;
88.                         }
89.                     }
90.                     break;
91.                 case OBSERVING:
92.                     break;
93.                 case FOLLOWING:
94.                     // 收到FOLLOWING通知
95.                     Vote resultFN = receivedFollowingNotification(recvset, outofelection, voteSet, n);
96.                     if (resultFN == null) {
97.                         break;
98.                     } else {
99.                         return resultFN;
100.                     }
101.                 case LEADING:
102.                     // 收到LEADING通知
103.                     Vote resultLN = receivedLeadingNotification(recvset, outofelection, voteSet, n);
104.                     if (resultLN == null) {
105.                         break;
106.                     } else {
107.                         return resultLN;
108.                     }
109.                 default:
110.                     break;
111.                 }
112.             } else {
113.                 // 略
114.             }
115.         }
116.         return null;
117.     } finally {
118.         // 略
119.     }
120. }

复制代码

receivedFollowingNotification方法

收到FOLLOWING状态notification。

1. private Vote receivedFollowingNotification(
2.     Map<Long, Vote> recvset, Map<Long, Vote> outofelection, SyncedLearnerTracker voteSet, Notification n) {
3.     // 也会将选票投给当前leader
4.     // 之后会进行quorum验证和leaderCheck验证
5.     if (n.electionEpoch == logicalclock.get()) {
6.         // 创建投票箱
7.         recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
8.         voteSet = getVoteTracker(
9.             recvset, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
10.         // acks>half和leaderCheck
11.         if (voteSet.hasAllQuorums() && checkLeader(recvset, n.leader, n.electionEpoch)) {
12.             // 更新节点状态
13.             setPeerState(n.leader, voteSet);
14.             Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
15.             leaveInstance(endVote);
16.             return endVote;
17.         }
18.     }
20.     // 当本节点较晚进入集群，集群已经有了leader时，会进入下面逻辑
21.     // 与前面的代码基本相同
22.     outofelection.put(n.sid, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
23.     voteSet = getVoteTracker(
24.         outofelection, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));
26.     if (voteSet.hasAllQuorums() && checkLeader(outofelection, n.leader, n.electionEpoch)) {
27.         synchronized (this) {
28.             logicalclock.set(n.electionEpoch);
29.             setPeerState(n.leader, voteSet);
30.         }
31.         Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch);
32.         leaveInstance(endVote);
33.         return endVote;
34.     }
36.     return null;
37. }

复制代码

receivedLeadingNotification方法

收到LEADING状态notification。

1. private Vote receivedLeadingNotification(Map<Long, Vote> recvset, Map<Long, Vote> outofelection,
2.                                          SyncedLearnerTracker voteSet, Notification n) {
3.     Vote result = receivedFollowingNotification(recvset, outofelection, voteSet, n);
4.     if (result == null) {
5.         if (self.getQuorumVerifier().getNeedOracle() && !self.getQuorumVerifier().askOracle()) {
6.             // 略
7.         } else {
8.             return null;
9.         }
10.     } else {
11.         return result;
12.     }
13. }

复制代码

QuorumPeer类

管理quorum协议，服务器可能处于以下三种状态：

• Leader选举 - 每个服务器将选出一个leader，最初都会选自己
  
• Follower节点 - 将与Leader同步并复制所有事务
  
• Leader节点 - 处理请求并将其转发给Follower节点，大多数Follower节点必须同步，该请求才能被提交
  

run方法main loop

run方法main loop判断当前peer状态，执行选举、lead、follow等逻辑：

1. public void run() {
2.     // 略
4.     try {
5.         // Main loop
6.         while (running) {
7.             switch (getPeerState()) {
8.             case LOOKING:
9.                 ServerMetrics.getMetrics().LOOKING_COUNT.add(1);
10.                 if (Boolean.getBoolean("readonlymode.enabled")) {
11.                     // 略
12.                 } else {
13.                     try {
14.                         reconfigFlagClear();
15.                         if (shuttingDownLE) {
16.                             shuttingDownLE = false;
17.                             startLeaderElection();
18.                         }
19.                         setCurrentVote(makeLEStrategy().lookForLeader());
20.                     } catch (Exception e) {
21.                         setPeerState(ServerState.LOOKING);
22.                     }
23.                 }
24.                 break;
25.             case OBSERVING:
26.                 try {
27.                     setObserver(makeObserver(logFactory));
28.                     observer.observeLeader();
29.                 } catch (Exception e) {
30.                 } finally {
31.                     observer.shutdown();
32.                     setObserver(null);
33.                     updateServerState();
35.                     // Add delay jitter before we switch to LOOKING
36.                     // state to reduce the load of ObserverMaster
37.                     if (isRunning()) {
38.                         Observer.waitForObserverElectionDelay();
39.                     }
40.                 }
41.                 break;
42.             case FOLLOWING:
43.                 try {
44.                     setFollower(makeFollower(logFactory));
45.                     follower.followLeader();
46.                 } catch (Exception e) {
47.                 } finally {
48.                     follower.shutdown();
49.                     setFollower(null);
50.                     updateServerState();
51.                 }
52.                 break;
53.             case LEADING:
54.                 try {
55.                     setLeader(makeLeader(logFactory));
56.                     leader.lead();
57.                     setLeader(null);
58.                 } catch (Exception e) {
59.                 } finally {
60.                     if (leader != null) {
61.                         leader.shutdown("Forcing shutdown");
62.                         setLeader(null);
63.                     }
64.                     updateServerState();
65.                 }
66.                 break;
67.             }
68.         }
69.     } finally {
70.         // 略
71.     }
72. }

复制代码

LOOKING分支

1. try {
2.     reconfigFlagClear();
3.     if (shuttingDownLE) {
4.         shuttingDownLE = false;
5.         startLeaderElection();
6.     }
7.     // 使用FastLeaderElection选举
8.     setCurrentVote(makeLEStrategy().lookForLeader());
9. } catch (Exception e) {
10.     setPeerState(ServerState.LOOKING); // 重置为LOOKING状态
11. }

复制代码

FOLLOWING分支

1. try {
2.     setFollower(makeFollower(logFactory));
3.     follower.followLeader(); // 启动follower
4. } catch (Exception e) {
5. } finally {
6.     follower.shutdown();
7.     setFollower(null);
8.     updateServerState(); // 更新服务状态
9. }

复制代码

创建Follower对象：

1. protected Follower makeFollower(FileTxnSnapLog logFactory) throws IOException {
2.     return new Follower(this, new FollowerZooKeeperServer(logFactory, this, this.zkDb));
3. }

复制代码

LEADING分支

1. try {
2.     setLeader(makeLeader(logFactory));
3.     leader.lead(); // 启动leader
4.     setLeader(null);
5. } catch (Exception e) {
6. } finally {
7.     if (leader != null) {
8.         leader.shutdown("Forcing shutdown");
9.         setLeader(null);
10.     }
11.     updateServerState(); // 更新服务状态
12. }

复制代码

创建Leader对象：

1. protected Leader makeLeader(FileTxnSnapLog logFactory) throws IOException, X509Exception {
2.     return new Leader(this, new LeaderZooKeeperServer(logFactory, this, this.zkDb));
3. }

复制代码

OBSERVING分支

1. try {
2.     setObserver(makeObserver(logFactory));
3.     observer.observeLeader();
4. } catch (Exception e) {
5. } finally {
6.     observer.shutdown();
7.     setObserver(null);
8.     updateServerState();
10.     // Add delay jitter before we switch to LOOKING
11.     // state to reduce the load of ObserverMaster
12.     if (isRunning()) {
13.         Observer.waitForObserverElectionDelay();
14.     }
15. }

复制代码

创建Observer对象：

1. protected Observer makeObserver(FileTxnSnapLog logFactory) throws IOException {
2.     return new Observer(this, new ObserverZooKeeperServer(logFactory, this, this.zkDb));
3. }

复制代码

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