zookeeper从3.4.0版本开始只保留了TCP版本的FastLeaderElection选举方法。
调用流程：QuorumPeer->looking状态(可以启动只读模式和阻塞模式)->lookForLeader
public Vote lookForLeader() throws InterruptedException { //... try { //用于选票归档 HashMap<Long, Vote> recvset = new HashMap<Long, Vote>(); HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>(); int notTimeout = finalizeWait; synchronized(this){ //logicalclock代表该当前机器的逻辑时钟（初始为0），每次进行一次leader选举，就会加一 logicalclock++; //初始化选票，投给自己，getInitLastLoggedZxid得到的是该服务器已经处理的事务的 //最大zxid，getPeerEpoch得到的是该服务器的选举轮次 updateProposal(getInitId(),getInitLastLoggedZxid(), getPeerEpoch()); } //初始化选票后发给所有服务器 sendNotifications(); while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){ //从 recvqueue获得来自所有机器的投票 Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); //如果没有获得投票 if(n == null){ //如果与其他服务器的连接仍然保持，重新发送投票 if(manager.haveDelivered()){ sendNotifications(); } else { //连接失效，重新建立连接。选举刚开始的时候就是这样建立连接的 manager.connectAll(); } //修改超时参数... } //否则处理选票 else if(self.getVotingView().containsKey(n.sid)) { switch (n.state) { case LOOKING: // 外部选票的逻辑时钟大于当前机器的逻辑时钟 if (n.electionEpoch > logicalclock) { logicalclock = n.electionEpoch; //更新当前机器的logicalclock（意味着当前机器逻辑时钟与集群一致了）并且清空接收的选票 recvset.clear(); //选票PK，外部投票更新，则变更自己的投票。 if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) { //变更选票 updateProposal(n.leader, n.zxid, n.peerEpoch); } else { //不变更选票 updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch()); } //敲黑板*******************不管变不变更选票，logicalclock都变了， //**********它会影响选票的electionEpoch，所以再将内部投票发送出去 sendNotifications(); } // 小于当前机器的逻辑时钟，直接丢弃 else if (n.electionEpoch < logicalclock) { break; } //等于当前机器的逻辑时钟，直接PK,外部选票PK赢了则更新内部投票并将内部投票发送出去 else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) { updateProposal(n.leader, n.zxid, n.peerEpoch); sendNotifications(); } //将收到的选票归档，<sid, 选票> //敲黑板*******************，recvset是HashMap<Long, Vote>，这里的key是投票发出方的服务器的id， //**************也就是说一个服务器只会在recvset保留一条记录，这就是为什么一台服务器可以多次发出投票 recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); //统计投票，决定是否终止投票,终止的条件是集群中过半的服务器认可了当前的内部投票，否则回到上面 //的while循环，继续循环 if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) { // 等一段时间（默认200ms）来确定是否有新的更优的投票 while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null){ if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)){ recvqueue.put(n); break; } } //没有收到更优的投票 if (n == null) { //设置状态，如果leader是自己，状态为Leading //如果leader是其他节点，状态可能为observing或者following self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch); //清空接收队列recvqueue，返回投票选举结果 leaveInstance(endVote); return endVote; } } break; case OBSERVING: break; //这两种情况发生在集群中本来就已经存在一个leader了，可能本台机器刚刚启动进入leader选举 //或刚因为网络延迟与leader断开然后重新进入选举 case FOLLOWING: case LEADING: //除了做出过半判断，同时还要检查leader是否给自己发送过投票信息，从投票信息中确认该 //leader是不是LEADING状态(防止出现时间差)。 //如果当前leader是真的有效，那一定有过半的机器（followers和leader）都会发来指向leade的选票 /* 同一轮投票选出leader，那么判断是不是半数以上的服务器都选举同一个leader， *如果是设置角色并退出选举 */ if(n.electionEpoch == logicalclock){ recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); if(ooePredicate(recvset, outofelection, n)) { self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch); leaveInstance(endVote); return endVote; } } /* 非同一轮次，例如宕机很久的机器重新启动/某个节点延迟很大变为looking，需要收集过半选票。*/ /*例如本机服务器是刚启动的，则我的logicalclock一定为1，假设这时候集群已经选好了leader了， *这时候本机一定收不到带looking状态的选票，所以无法更新自己的logicalclock，因此我们将这种 *的选票不在放入recvset，而是放入outofelection，当收到过半选票时，更新本机的logicalclock *与集群同步，大家都在同一逻辑时钟并设置好对应角色退出选举*/ outofelection.put(n.sid, new Vote(n.version, n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state)); if(ooePredicate(outofelection, outofelection, n)) { synchronized(this){ logicalclock = n.electionEpoch; self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); } Vote endVote = new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch); leaveInstance(endVote); return endVote; } break; default: break; } } else { LOG.warn("Ignoring notification from non-cluster member " + n.sid); } } return null; } }选票vote
初始化选票（对应上述lookforleader的初始化选票）
(sid, LastLoggedZxid, currentEpoch)LastLoggedZxid为机器上一次处理的最后一个事务的zxid(包括提交，未提交)updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
id 当前服务器自身的sid zxid 当前服务器最新的zxid值 electionEpoch 当前服务器的逻辑时钟 peerEpoch state looking 接收到新的选票后，从依次按以下几个层次判断 选票状态选票轮次选票变更规则变更选票的3条规则 return ((newEpoch > curEpoch) || ((newEpoch == curEpoch) && ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId))))); 选择peerEpoch更高（被推举的服务器的逻辑时钟更高的）peerEpoch相同，选择zxid更高的zxid相同，选择sid更大的每台服务器启动的时候，都会启动一个QuorumCnxManager，负责各台服务器之间的leader选举过程中的网络通信。
QuorumCnxManager这个类内部维护一系列的队列，用于保存接收到的，待发送的消息，以及消息的发送器。除了接收队列外，这里提到的所有队列都按SID分组形成队列集合。
/*消息接收队列只有一个*/ public final ArrayBlockingQueue<Message> recvQueue; /*消息发送队列按SID分组，分别为集群中每台机器分配一个单独队列*/ final ConcurrentHashMap<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap; /*发送器集合，按SID分组,每个SendWorker消息发送器对应一台远程zookeeper服务器，负责从对应的发送队列取出消息发送*/ final ConcurrentHashMap<Long, SendWorker> senderWorkerMap; /*为每个SID保留最近发送过的一个消息*/ final ConcurrentHashMap<Long, ByteBuffer> lastMessageSent;为能互相投票，zookeeper集群的所有机器都需要两两建立起网络连接。QuorumCnxManager启动时，会创建一个ServerSocket来监听Leader选举的通信端口（默认端口是3888）。开启端口监听后，就能接收其他服务器的“创建连接”请求。
QuorumPeer.java Election createElectionAlgorithm(int electionAlgorithm){ ... case 3: qcm = createCnxnManager(); QuorumCnxManager.Listener listener = qcm.listener; if(listener != null){ listener.start(); le = new FastLeaderElection(this, qcm); } ... } public class Listener extends ZooKeeperThread { volatile ServerSocket ss = null; public void run() { int numRetries = 0; InetSocketAddress addr; while((!shutdown) && (numRetries < 3)){ try { ss = new ServerSocket(); ss.setReuseAddress(true); ... ss.bind(addr); while (!shutdown) { Socket client = ss.accept(); setSockOpts(client); ... /*接收到其他服务器TCP连接请求时交由receiveConnection处理*/ if (quorumSaslAuthEnabled) { receiveConnectionAsync(client); } else { receiveConnection(client); } numRetries = 0; } } ... }为了防止两台服务器有重复链接，zookeeper定义了规则，只能sid大的去连接sid小的。如果sid小的连接了sid大的，在连接处理程序中会断掉这条连接，然后重新发起连接。
public void receiveConnection(final Socket sock) { DataInputStream din = null; try { din = new DataInputStream( new BufferedInputStream(sock.getInputStream())); handleConnection(sock, din); } ... } private void handleConnection(Socket sock, DataInputStream din) throws IOException { Long sid = null; try { // 读取远端服务器sid sid = din.readLong(); if (sid < 0) { sid = din.readLong(); .... } .... //如果对方id比我小，则关闭连接，只允许大id的server连接小id的server if (sid < this.mySid) { SendWorker sw = senderWorkerMap.get(sid); if (sw != null) { sw.finish(); } closeSocket(sock); connectOne(sid); //关闭连接后，主动去连对面 } //如果对方id比我大，允许连接，并初始化单独的IO线程 else { SendWorker sw = new SendWorker(sock, sid); RecvWorker rw = new RecvWorker(sock, din, sid, sw); sw.setRecv(rw); SendWorker vsw = senderWorkerMap.get(sid); if(vsw != null) vsw.finish(); senderWorkerMap.put(sid, sw); if (!queueSendMap.containsKey(sid)) { queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>( SEND_CAPACITY)); } //这一点我有个疑问，可以看出上面会先用queueSendMap.containsKey(sid)判断原有sid //是不是已经有对应的消息发送队列，如果没有创建新的，意思是如果有就用旧的消息发送队 //列，我的疑问就是这时旧的消息发送队列可能会包含上轮选举的旧消息，为什么这里不对它清 //空呢，不清空会把旧的选票信息发给对应的sid服务器，虽然对选举结果没啥影响，但感觉清空 //队列效率更高 sw.start(); rw.start(); return; }一旦 建立起连接，就会根据远程服务器的SID来创建相应的消息发送器SendWorker和消息接收器RecvWorker,并启动他们。
主动发起连接的server的IO线程初始化
//主动去连对面 synchronized public void connectOne(long sid){ Socket sock = new Socket(); setSockOpts(sock); sock.connect(view.get(sid).electionAddr, cnxTO); ... initiateConnection(sock, sid); ... } public void initiateConnection(final Socket sock, final Long sid) { ... startConnection(sock, sid); } private boolean startConnection(Socket sock, Long sid) throws IOException { DataOutputStream dout = null; DataInputStream din = null; try { //先发一个server id，让对面和自己的sid作比较以决定是否关闭连接，只允 //许sid大的连小的 dout = new DataOutputStream(sock.getOutputStream()); dout.writeLong(this.mySid); dout.flush(); din = new DataInputStream( new BufferedInputStream(sock.getInputStream())); } ..... // If lost the challenge, then drop the new connection //如果对方id比自己大，则关闭连接，这样导致的结果就是大id的server才会去连接小id //的server，避免连接浪费 ，对方收到连接后会主动再连过来 if (sid > self.getId()) { LOG.info("Have smaller server identifier, so dropping the " + "connection: (" + sid + ", " + self.getId() + ")"); closeSocket(sock); // Otherwise proceed with the connection } //如果对方id比自己小，则保持连接，并初始化单独的发送和接受线程 else { SendWorker sw = new SendWorker(sock, sid); RecvWorker rw = new RecvWorker(sock, sid, sw); sw.setRecv(rw); SendWorker vsw = senderWorkerMap.get(sid); if(vsw != null) vsw.finish(); senderWorkerMap.put(sid, sw); if (!queueSendMap.containsKey(sid)) { queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>( SEND_CAPACITY)); } //这个疑问与上面疑问一样，旧的消息发送队列可能会包含上轮选举的旧消息，为 //什么这里不对它清空呢 sw.start(); rw.start(); return true; } return false; }IO发送线程SendWorker启动，开始发送选举消息
class SendWorker extends ZooKeeperThread { Long sid; Socket sock; RecvWorker recvWorker; volatile boolean running = true; DataOutputStream dout; public void run() { try { while (running && !shutdown && sock != null) { ByteBuffer b = null; try { //每个server一个发送队列 ArrayBlockingQueue<ByteBuffer> bq = queueSendMap .get(sid); if (bq != null) { //拿消息 b = pollSendQueue(bq, 1000, TimeUnit.MILLISECONDS); } else { LOG.error("No queue of incoming messages for " + "server " + sid); break; //如果消息发送队列没消息则关闭消息发送器，一般不会出现这种情况，因为 FastLeaderElection有如下代码， //haveDelivered就是检验各个消息发送队列是否都为空，空则重复发送刚才的提议的选票给消息发送队列 /* if(manager.haveDelivered()){ sendNotifications(); } */ } if(b != null){ //发消息 lastMessageSent.put(sid, b); send(b); } } } } this.finish(); ...... /** * Check if all queues are empty, indicating that all messages have been delivered. */ boolean haveDelivered() { for (ArrayBlockingQueue<ByteBuffer> queue : queueSendMap.values()) { LOG.debug("Queue size: " + queue.size()); if (queue.size() == 0) { return true; } } return false; }这个时候，其他机器通过IO线程RecvWorker收到消息
class RecvWorker extends ZooKeeperThread { Long sid; Socket sock; volatile boolean running = true; final DataInputStream din; final SendWorker sw; public void run() { threadCnt.incrementAndGet(); try { while (running && !shutdown && sock != null) { //包的长度 int length = din.readInt(); if (length <= 0 || length > PACKETMAXSIZE) { throw new IOException( "Received packet with invalid packet: " + length); } //读到内存 byte[] msgArray = new byte[length]; din.readFully(msgArray, 0, length); ByteBuffer message = ByteBuffer.wrap(msgArray); //添加到接收队列，后续业务层()的接收线程WorkerReceiver会来拿消息 addToRecvQueue(new Message(message.duplicate(), sid)); } ...... }1.server启动时默认选举自己，并向整个集群广播 2.收到消息时，通过3层判断：逻辑时钟，zxid，server id大小判断是否同意对方，如果同意，则修改自己的选票，并向集群广播 3.QuorumCnxManager负责IO处理，每2个server建立一个连接，只允许id大的server连id小的server，每个server启动单独的读写线程处理，使用阻塞IO 4.默认超过半数机器同意时，则选举成功，修改自身状态为LEADING或FOLLOWING 5.Obserer机器不参与选举
updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
获得的epoch就是选票所推举的服务器的逻辑时钟，它其实就是zxid的前半部。
private long getPeerEpoch(){ if(self.getLearnerType() == LearnerType.PARTICIPANT) try { return self.getCurrentEpoch(); } public long getCurrentEpoch() throws IOException { if (currentEpoch == -1) { //刚启动时，会从名叫currentEpoch的文件读取获得 currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME); } return currentEpoch; } public void setCurrentEpoch(long e) throws IOException { currentEpoch = e; writeLongToFile(CURRENT_EPOCH_FILENAME, e); //会把currentEpoch记入currentEpoch文件 }下面我们看看哪里会调用setCurrentEpoch
Leader.java void lead() { long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch()); .... waitForEpochAck(self.getId(), leaderStateSummary); self.setCurrentEpoch(epoch); }上面代码是准leader在lead时先生成新的new epoch且获得过半机器的ack，大家将一起用new epoch作为新的选举轮次，也就是zxid的前面部分。
QuorumPeer.java private void loadDataBase() { long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid; long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid); currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME); if (epochOfZxid > currentEpoch && updating.exists()) { setCurrentEpoch(epochOfZxid);上面是加载内存数据库时从日志文件中分别读取上次最后处理的zxid以及上次的currentEpoch，因为currentEpoch代表的就是当前leader的选举轮次，它是不应该小于zxid的前半部的，如果小于，就用lastProcessedZxid的epoch部分更新。
Learner.java protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException{ .... case Leader.NEWLEADER: self.setCurrentEpoch(newEpoch);可以看出learner在与leader同步时会将epoch设置为与leader相同的epoch
