★ FEATURED ARTICLE 1. 项目概述1.1 核心需求解析熟悉 ZooKeeper 源码的人都知道Leader 选举是整个 ZooKeeper 最核心、也最绕的一段逻辑。有人形容看选举源码像看悬疑小说——变量多、线程多、状态切换复杂一个不留神就跟丢了。我之前在系列文章里已经带大家走过了 ZooKeeper 启动的完整流程从配置文件加载到 QuorumPeer 主线程启动再到选举算法的初始化。这次咱们把镜头对准选举中最关键的一条链路投票的发送与响应。这条链路贯穿了 QuorumCnxManager 的队列体系、SendWorker/RecvWorker 两个后台线程以及 FastLeaderElection 的轮询处理逻辑。标题里虽然写的是启动四但这一篇其实可以独立阅读。只要你理解三个基本概念——集群节点通过投票来决定谁是 Leader、投票需要相互交换、节点根据收到的投票调整自己的选择——就能跟上这篇文章的节奏。这篇文章适合三类读者一是准备面试大厂中间件岗位、需要手撕 ZooKeeper 选举源码的开发者二是在生产环境遇到选举卡死、选主超时等问题、想从源码层面定位根因的运维工程师三是纯粹想提升源码阅读能力的 Java 后端。我会严格控制节奏先讲整体链路再逐段拆解最后给出排查经验。1.2 技术选型与阅读路径在深入源码之前先明确我们关注的核心类类名职责QuorumCnxManager选举通信层负责维护节点间的 TCP 连接、消息队列SendWorker发送线程从发送队列取消息并写往对端RecvWorker接收线程从 TCP 连接读消息并放入 recvQueueFastLeaderElection选举算法核心包含 vote 比较、状态切换、轮询处理Vote投票数据结构包含 id、zxid、epoch 等关键字段阅读路径建议按运行时序先看 QuorumCnxManager 的构造与连接建立再看 SendWorker/RecvWorker 的循环逻辑然后进入 FastLeaderElection 的 sendNotifications 和 处理收到的消息最后看选举完成条件。2. 内容整体设计与思路拆解2.1 为什么先要理解 QuorumCnxManager很多人第一次读选举源码会犯一个错误直接扎进 FastLeaderElection 的 loop 方法里被各种 if-else 绕晕。其实 FastLeaderElection 本身并不负责网络传输它只是把我要选谁封装成 Notification 对象至于这个 Notification 怎么送到对端节点、对端节点的回复怎么收回来完全是 QuorumCnxManager 的事。打个比方FastLeaderElection 是决策层QuorumCnxManager 是通信层。决策层只需要说我决定投给节点 3通信层负责把这个决定可靠地送达。如果通信层出了问题比如连接断开会话超时决策层再聪明也选不出 Leader。所以阅读源码的顺序必须是先通信层后决策层。QuorumCnxManager 的核心数据结构是ArrayBlockingQueueMessage queueSendMap这是一个以对端节点 sid 为 key 的发送队列映射表。每个节点在内存里维护了我要发给其他所有节点的消息队列。另一个关键结构是recvQueue所有从对端收到的消息都先进这个队列再由 FastLeaderElection 的轮询线程统一消费。这个设计与我们日常开发中的生产-消费模型非常相似。发送方把消息丢进队列就返回不关心对端什么时候处理接收方从队列里拉取消息逐个处理。队列解耦了网络 I/O 与业务逻辑这也是 ZooKeeper 在极端网络条件下依然能保证选举不彻底卡死的重要原因。2.2 从启动流程看选举入口回顾一下启动链路QuorumPeerMain 启动后加载配置创建 QuorumPeer 实例调用 start() 方法。QuorumPeer 的 start 方法中有一个关键分支——如果没有 Leader就进入 LOOKING 状态触发startLeaderElection()。startLeaderElection里第一步是创建 QuorumCnxManager传入self.getId()当前节点 sid、self.getVote()当前投票、self.getElectionAddress()选举端口等参数。这里有个细节QuorumCnxManager 构造时会尝试与集群中所有 sid 小于自己的节点建立连接大于自己的节点则等待对方主动连接。这种一半主动一半被动的设计是为了避免两个节点同时向对方发起 TCP 连接造成连接资源浪费。然后创建 FastLeaderElection 实例把 QuorumCnxManager 传进去。FastLeaderElection 构造时会启动两个后台线程WorkerReceiver和WorkerSender。前者负责从 recvQueue 取消息并调用receivedNotification后者负责把本地投票封装成 Notification 并调用sendqueue。这两个线程是选举算法的心脏后面会详细展开。3. 核心细节解析与实操要点3.1 Vote 对象内部结构与关键字段Vote 是选举中最基本的选票数据结构我先把字段列出来再逐个解释public class Vote { private final long id; // 被投票节点的 sid private final long zxid; // 被投票节点的事务 id private final long electionEpoch; // 选举轮次 private final long peerEpoch; // 对端节点的 epoch private final long round; // 当前轮次 private final boolean notifier; // 是否为通知型投票 }很多文章会忽略peerEpoch和electionEpoch的区别导致读者对后面的逻辑判断一头雾水。我用一个场景解释假设集群有三个节点sid 分别为 1、2、3。节点 1 和节点 2 开始新一轮选举这时集群的 electionEpoch 是 1。节点 1 的数据版本比较新它的 zxid 是 0x100000001节点 2 的数据略旧zxid 是 0x100000000。节点 1 发起投票投自己Vote 中 id1, zxid0x100000001, electionEpoch1。节点 2 收到这票后会先对比 electionEpoch。如果对方选举轮次大于自己说明对方已经进入新一轮自己也要跟着更新 epoch。这个机制可以防止旧轮次的迟到消息干扰新轮次。在源码中Vote的hashCode和equals也很讲究。equals方法比较的是 id、zxid 和 electionEpoch没有比较 peerEpoch 和 notifier 字段。这意味着两个在不同时间点构造的 Vote只要核心三个字段一致就被认为是相同投票。这个设计直接影响 FastLeaderElection 中是否收到足够多的相同投票的判断逻辑。3.2 Notification 消息封装与标识规则FastLeaderElection 中定义了内部类Notification它是对 Vote 的网络传输包装。字段比 Vote 多一些public static class Notification { private final long leader; // 提议的 leader sid private final long zxid; private final long electionEpoch; private final long state; // 发送方当前状态 private final long sid; // 发送方 sid private final long peerEpoch; private final byte[] data; // 附加数据 }注意state字段它表示发送方当前的状态LOOKING、FOLLOWING、LEADING 或 OBSERVING。这个字段非常关键。在选举完成之后节点仍然会收到其他节点发来的消息如果状态不是 LOOKING接收方会走另一条处理逻辑——把对方当作新 Leader 或已就绪的节点而不会进入选票比较流程。消息在网络上传输时不是直接发 Java 对象而是通过ByteBuffer序列化。ZooKeeper 定义了LEADERINFO、CEPOCH、NEWLEADER、ACK、FOLLOWINGINFO、LOOKINGINFO等消息类型。选举阶段主要用LOOKINGINFO类型消息体里写入 proposal 相关的数据。这块逻辑集中在FastLeaderElection的sendNotifications方法中void sendNotifications() { Notification n new Notification(); n.leader self.getVote().getId(); n.zxid self.getLastLoggedZxid(); n.electionEpoch self.getElectionEpoch(); n.state self.getPeerState(); n.sid self.getId(); n.peerEpoch self.getPeerEpoch(); // 序列化并发送 sendqueue.offer(n); }这里有个容易忽略的点n.zxid取的是self.getLastLoggedZxid()这是当前节点本地日志中最后一条事务的 zxid。zxid 越大代表数据越新。投票时优先比较 zxid再比较 sid这就是 ZooKeeper 选举数据新者优先原则的代码体现。3.3 队列体系设计从 sendqueue 到 queueSendMapFastLeaderElection 维护了一个ArrayBlockingQueueNotification sendqueue这是待发送消息的入口队列。WorkerSender 线程从这个队列取消息然后根据目标 sid 放入 QuorumCnxManager 的 queueSendMap 对应队列中。class WorkerSender extends ZooKeeperThread { public void run() { while (!stop) { Notification n sendqueue.poll(3000, TimeUnit.MILLISECONDS); if (n null) continue; // 根据 n.sid 找到目标节点的发送队列 QuorumCnxManager.Message message new QuorumCnxManager.Message( ByteBuffer.wrap(serialize(n)), n.sid); if (manager ! null) { manager.toSend(n.sid, message); } } } }QuorumCnxManager.toSend(long sid, Message message)的逻辑是这样的先检查与目标 sid 的连接是否建立如果没建立尝试建立连接然后从 queueSendMap 中取出目标对应的队列把消息放入。队列是用ArrayBlockingQueue实现的容量默认是Integer.MAX_VALUE所以正常情况下不会因为队列满而阻塞。这个两级队列的设计值得借鉴。第一级 sendqueue 是逻辑队列FastLeaderElection 只关心把消息交给 WorkerSender不关心网络细节第二级 queueSendMap 是物理队列按目标节点隔离。好处显而易见一个节点网络慢或断开只会影响它自己对应的队列不会阻塞其他节点的消息发送。4. 实操过程与核心环节实现4.1 连接建立主动连接与被动接收在 QuorumCnxManager 构造时会调用connectAll()方法尝试与所有比自己 sid 小的节点建立连接public void connectAll() { // 遍历所有参与投票的节点 for (Long sid : self.getCurrentVotingView().keySet()) { if (sid self.getId()) continue; if (sid self.getId()) { // 主动连接 sid 比自己小的节点 connectOne(sid); } // sid 比自己大的节点等待对方来连接 } }为什么选择主动连接小 sid、被动等待大 sid的策略想象一下如果不加这个约束节点 1 和节点 2 同时 connect 对方就会同时建立两条 TCP 连接。ZooKeeper 虽然不会立刻报错但会造成资源浪费和状态混乱。有了 sid 大小约束每对节点之间最多只有一条主动连接另一侧通过监听端口接受连接天然避免了重复建连。connectOne方法内部会检查listenerSemaphore信号量。这个信号量的存在是为了限制并发连接数量防止大量节点同时启动时瞬间建立海量连接导致线程资源耗尽。如果信号量获取失败说明系统当前连接压力过大QuorumCnxManager 会等一会再重试。连接建立后会为每个对端节点启动一对线程SendWorker和RecvWorker。这两个线程和一个ArrayBlockingQueueMessage绑定共同组成一条完整的双向通信管道。4.2 SendWorker 的发送循环与流控细节先看 SendWorker 的核心循环逻辑。这里我简化了源码保留关键点class SendWorker extends ZooKeeperThread { Override public void run() { while (!stop !isInterrupted()) { try { Message m queue.poll(3000, TimeUnit.MILLISECONDS); if (m null) continue; if (m instanceof Message) { // 写入 DataOutput 流 dout.writeInt(m.getBuffer().remaining()); dout.write(m.getBuffer().array()); dout.flush(); } } catch (IOException e) { // 发送失败处理连接断开 handleSocketFailure(); } } } }注意writeInt写的是消息体的长度服务端接收时先读长度再读对应长度的字节。这是最经典的长度前缀 消息体编解码方式也是 ZooKeeper 自定义协议的基础。发送线程每 3 秒轮询一次队列如果队列为空就空转一轮。这样设计的好处是能及时发现线程中断和连接异常而不是无限阻塞在队列上。这里有一个非常关键的细节当handleSocketFailure被触发时SendWorker 会把自己从 QuorumCnxManager 的senderWorkerMap中移除并关闭对应的 socket。但queueSendMap 中对应目标节点的队列不会删除。也就是说失败期间积累的待发送消息仍然保留在内存中等连接恢复后继续发送。这个设计能在网络抖动时保证消息不丢失。不过也要注意如果连接长时间无法恢复队列中积压的消息会越来越多内存压力不可小觑。生产环境如果看到某个节点的内存持续上涨排查方向之一就是看看是否有对端节点连接异常导致消息积压。4.3 RecvWorker 的接收循环与消息解析RecvWorker 的逻辑相对简单但也最容易出问题class RecvWorker extends ZooKeeperThread { Override public void run() { while (!stop !isInterrupted()) { try { int length din.readInt(); if (length 0 || length maxBuffer) { throw new IOException(Invalid length: length); } byte[] msgArray new byte[length]; din.readFully(msgArray); Message msg new Message(ByteBuffer.wrap(msgArray), sid); manager.recvQueue.offer(msg); } catch (IOException e) { handleSocketFailure(); } } } }我特意加了一个length的合法性检查。在实际源码中确实有maxBuffer限制默认 0xFFFFF 即 1MB。如果收到的长度值异常直接抛异常并断开连接防止恶意或异常消息导致内存溢出。这个防御性编程思路在自研 RPC 框架时同样适用。RecvWorker 收到消息后直接offer到recvQueue不做任何解析。这意味着 FastLeaderElection 的 WorkerReceiver 才是真正决定消息该走哪条逻辑的路由器。4.4 WorkerReceiver 的处理分支详解WorkerReceiver 是选举算法里最绕的部分因为它要根据收到的 Notification 中的state字段和当前节点自身的状态决定是进入选票比较还是进入确认逻辑。class WorkerReceiver extends ZooKeeperThread { public void run() { while (!stop) { // 从 recvQueue 获取消息 QuorumCnxManager.Message message manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS); if (message null) continue; // 反序列化为 Notification Notification n deserialize(message.getBuffer()); // 根据发送方状态和本节点状态分派 switch (n.state) { case LOOKING: { // 如果自己也处于 LOOKING加入选票比较 // 如果自己已经不是 LOOKING记录到 already_notified break; } case FOLLOWING: case LEADING: { // 处理新的 leader 信息 break; } } } } }当本节点处于 LOOKING 状态收到对端也是 LOOKING 的消息时会把消息转交给receivedNotification方法由 electionAlgorithm 处理。如果本节点已经选出了 Leader不再 LOOKING但收到了其他节点的 LOOKING 通知说明这个节点可能刚启动或刚发生崩溃恢复还在找 Leader。这时本节点会把自己的当前 Leader 信息再次广播回去帮助对方快速定位 Leader。这里有个很隐蔽的设计WorkerReceiver 有 setter 方法setVote、setEpoch、setZxid这些 setter 在消息处理中途会被外部调用。这不是并发 bug而是为了在本节点状态已经变化时及时用最新的投票信息回应对端。源码阅读时看到方法内数据突然变化不要慌先看看是不是有线程在背后更新状态。4.5 FastLeaderElection 的选票比较逻辑进入receivedNotification后真正核心的逻辑是对投票的合法性判断与比较。看这段代码前先记住 ZooKeeper 选举的四个优先比较条件优先级从高到低选举轮次 electionEpoch越大越新只有大的才能覆盖小的peerEpoch节点所在 epochzxid数据版本越大代表数据越新sid节点编号越大优先级越高源码中totalOrderPredicate方法实现了这个比较protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) { if (self.getQuorumVerifier().getWeight(newId) self.getQuorumVerifier().getWeight(curId)) { return true; } if (self.getQuorumVerifier().getWeight(newId) self.getQuorumVerifier().getWeight(curId)) { if (newEpoch curEpoch) return true; if (newEpoch curEpoch) { if (newZxid curZxid) return true; if (newZxid curZxid) { if (newId curId) return true; } } } return false; }注意这段代码体现了QuorumVerifier的权重参与比较。默认权重是 1所以比较退化为 epoch→zxid→sid 的顺序。但如果配置了group和weight权重更大的节点在选举中会被优先选中这是 ZooKeeper 3.5 之后引入的能力。每个节点在收到一条合法的选票后会更新自己的proposedLeader、proposedZxid、proposedEpoch然后调用sendNotifications()把更新后的投票广播出去。这个过程会一直持续直到某个节点的投票被集群中超过半数的节点认可。4.6 选举完成的判定条件选举完成判定是FastLeaderElection.termPredicate方法承担的。它检查recvset中是否有某个节点获得了超过半数的投票。recvset是一个HashMapLong, Votekey 是投票方的 sidvalue 是该方最后一次发出的投票。每收到一条合法选票就把发送方 sid 对应的 Vote 更新到 recvset 中。注意更新的是这条选票投给谁而不是这条选票由谁投出。所以当遍历 recvset 统计票数时是在统计有多少节点投给了同一个目标节点。protected boolean termPredicate(HashMapLong, Vote votes, Vote vote) { HashSetLong set new HashSetLong(); for (Map.EntryLong, Vote entry : votes.entrySet()) { if (vote.equals(entry.getValue())) { set.add(entry.getKey()); } } return self.getQuorumVerifier().containsQuorum(set); }胜出条件就是投票给自己的节点数达到法定数量。对默认的 3 节点集群法定数量是 2对 5 节点集群法定数量是 3。一旦某个节点在 recvset 中获得了法定数量的支持就退出循环执行leading()或following()状态切换。这里有个细节recvset 中的 Vote 不仅要 equal还需要对方的选举轮次与当前一致。由于equals不比较 peerEpoch如果出现跨轮次的脏数据理论上可能出现误判。但 ZooKeeper 在每轮选举开始时会clearrecvset加上 electionEpoch 的比较门槛实际不会出问题。不过这提醒我们在实现类似选举算法时轮次隔离是第一优先级否则旧轮次数据会污染新轮次投票。5. 常见问题与排查技巧实录5.1 发送队列积压选举持久不结束现象集群中某个节点一直处于 LOOKING 状态日志中不断打印Notification但选举就是不结束。用jstack能看到 SendWorker 线程阻塞在ArrayBlockingQueue.put或offer上或者 RecvWorker 线程在readFully处卡住。排查思路先检查目标节点的连接状态。queueSendMap中积压大量消息往往意味着对端 socket 已断开但 SendWorker 没有及时清理。查看是否触发了handleSocketFailure。这个方法会把连接从 senderWorkerMap 移除但如果queueSendMap中对应队列里的消息没有清除且连接重建失败就会持续积压。检查文件描述符数量。ZooKeeper 节点数多时每个节点与其他节点建立连接会消耗大量 fd如果操作系统限制过小连接可能反复建立失败。我踩过的一个坑某次压测环境同时启动 7 个节点由于maxClientCnxns和操作系统 ulimit 设置不当大量连接被拒绝导致节点之间无法交换投票选举迟迟不收敛。解决方式是提高 ulimit、缩小snapCount、分批启动节点让连接建立过程有节奏地进行。这个问题在自研分布式系统时也经常遇到建议在部署脚本里就提前检查 fd 数量和线程数限制。5.2 选票迟到导致选举反复现象集群出现 leader 切换后原来的 Leader 又收到旧轮次的 LOOKING 消息导致它退出 LEADING 状态重新选举。这是由WorkerReceiver的消息处理分支决定的。当节点处于 LEADING 或 FOLLOWING 状态时如果收到 LOOKING 消息会调用sendNotifications通知对方自己当前的状态帮助对方快速加入。但由于网络延迟这条通知的到达时间可能晚于对方已经完成的第二轮选举导致双方出现短暂的误会。排查方法在日志中关注New election epoch和Updating proposal关键字如果频繁出现说明集群中至少有一个节点在反复进入 LOOKING 状态。重点检查这个节点的磁盘 I/O、事务日志写入延迟因为这些会导致它的 zxid 落后于其他节点从而反复触发比较逻辑。实际上ZooKeeper 的设计已经考虑到了这种情况——每个节点在收到新轮次消息时会递增自己的 electionEpoch并清空 recvset。只要所有节点都能正确推进 epoch旧消息的影响会被控制在有限范围内。如果生产环境频繁出现选举反复更可能的原因是节点间时钟偏移过大或网络分区持续存在建议先排查硬件和网络条件。5.3 投票一致但无法达到法定数量现象打开日志看到所有节点都已经投票给了同一个节点但termPredicate一直不满足。关键在QuorumVerifier.containsQuorum方法。默认情况下containsQuorum(SetLong set)检查的是 set 的大小是否大于总节点数的一半。但如果配置了group该方法会先判断每个 group 内是否有节点投票再统计 group 层面的支持情况。我遇到的一个案例测试环境把 5 个节点配置成了 3 个 group其中一个 group 有 3 个节点另两个 group 各 1 个节点。因为配置不当导致法定数量计算方式与预期不同某一组投票始终无法满足要求。排查方式很简单通过 JMX 或 admin 接口查看QuorumVerifier的配置确认 group 和 weight 是否符合预期。如果不涉及多数据中心部署建议不要自定义 group保持默认的简单多数即可。6. 实操总结与后续扩展建议6.1 我的代码阅读路径如果你也想手撕 ZooKeeper 选举源码我建议的路径是先读QuorumCnxManager的构造方法和connectAll搞懂连接建立机制。再读SendWorker和RecvWorker的 run 方法理解消息如何进出队列。进入FastLeaderElection先看sendNotifications和receivedNotification再追totalOrderPredicate和termPredicate。最后回头看QuorumPeer.run的 LOOKING 分支把选举和状态切换串起来。按这个顺序读下来你对 ZooKeeper 选举的理解会非常立体而不只是零散地记住几个类名。6.2 对自研分布式系统的启发ZooKeeper 选举的通信层设计非常有参考价值队列解耦两个队列sendqueue 和 queueSendMap隔离了业务逻辑与网络细节。半连接策略按 sid 大小分配主动/被动连接避免建连冲突。失败不丢消息连接断开但队列保留恢复后继续发送。轮次隔离每次选举递增 epoch清洗旧数据防止脏投票干扰新选举。这些设计思路可以直接借鉴到自研的 Raft/Paxos 实现、分布式锁组件或者配置中心中。我在自己的一个模拟项目中就采用了类似的双层队列 按节点隔离模式在高并发写入场景下表现稳定排查问题也比单队列模型直观得多。6.3 下一步可以看什么本篇讲的是选举过程中的投票发送和响应。下一篇可以接着看选举完成后 Leader 如何向 Followers 同步数据也就是LearnerHandler和Learner的交互逻辑。这条链路比选举更复杂涉及 zxid 比较、事务批处理、snapshot 传输等大量细节但有了本篇建立的通信模型基础读起来会顺畅很多。我个人阅读源码的习惯是把关键流程画成时序图然后用 debug 日志去验证自己的理解是否与源码一致。事实证明带着问题去读代码比从头到尾逐行读效率高得多。如果你读完这篇有疑问或者发现某些细节与我描述的预期不符建议把对应版本的源码打开对照着看——源码永远是最好的老师。
阅读完成 · 觉得有帮助？
