欢迎大家关注 github.com/hsfxuebao ，希望对大家有所帮助，要是觉得可以的话麻烦给点一下Star哈
在前面一篇已经分析过了FLE的原理流程以及通信结构，接下来便详细分析一下ZK集群在建立通信结构源码层面的操作流程，在进行接下来的源码通信流程前需要对这个流程有一个大概的认识，否则很容易分析到一半便开始失去思路。本篇只分析创建集群通信对的流程，具体的选举源码留到下篇再来分析。
本篇是对上一篇的补充，建议在看这次的源码分析时可以对照着上一篇的流程图来看，更容易跳出代码知道在流程中的作用，上一篇的链接：FLE(FastLeaderElection)算法集群选举通信原理及流程结构(类解读)
注：本篇基于ZK版本3.7分析的。
1.源码分析
本次的源码分析步骤和以前不太一样，由于ZK集群选举时涉及三种角色：Leader、Follower和Observer，三种不同角色的源码流程是不一样的，因此本次源码将会以Leader和Follower两种角色的源码流程分析，从开始选举流程通用的流程开始，当各个机器有不同的流程时再分开依次分析，Observer由于不参与选举，因此本次源码分析忽略。
本次源码分析假设ZK集群有三台机器：
A机器：myid=1，启动时间最早；
B机器：myid=3，启动时间在A之后；
C机器：myid=5，启动时间最后。
接下来将会以这三台机器为例开始逐步分析ZK的选举流程。构建集群内部通信对源码解析整体流程图：
2. QuorumPeer对象发起投票
在看过前面有一篇的启动组价及流程可以知道，ZK集群内的每台机器都会有一个QuorumPeer集群对象，此对象是个线程对象，用来监听本机的状态：1、选举流程状态；2、确认角色后的数据同步流程状态，起到一个统筹兼顾的作用。关键源码如下：
public class QuorumPeer extends ZooKeeperThread implements QuorumStats.Provider { // 集群对象会含有三种不同的角色对象，如果机器在选举时被表明了是什么角色时 // 对应的对象将会被初始化，代表着本机器的角色，执行相应的操作 public Follower follower; public Leader leader; public Observer observer; // 代表着本机的当前投票 volatile private Vote currentVote; @Override public void run() { // 源码中这里有个流程是用来注册JMX对象的，这里和选举流程无关因此忽略 try { // 正式开始ZK集群执行流程，这里会有三种情况： // 1、如果在选举流程，peerState将一直会是LOOLING，直到集群选举出 // Leader；2、当选出Leader后，本机器的peerState将会变成对应的状态 // 直到Leader宕机不得不选举出新的Leader；3、每一次新的轮询都代表 // 着本机的角色发生了改变，执行的作用也发生改变。 while (running) { switch (getPeerState()) { // 代表本机正在进行选举流程 case LOOKING: // 本机是否开启只读模式，有兴趣的可以去看下，本次只分析 // 普通正常的流程 if (Boolean.getBoolean("readonlymode.enabled")) { // 忽略 ... } else { try { // 选举前先把之前的投票清空，以免对选举流程产生误导 setBCVote(null); // 设置本次投票结果，lookForLeader()方法里面将会 // 一直轮询和集群内的机器进行通信，直到选举出新的 // Leader或者发生了异常情况 setCurrentVote( makeLEStrategy().lookForLeader()); } catch (Exception e) { // 如果发生了异常情况则设置本机的状态为选举中 // 以便进入下一次选举流程 setPeerState(ServerState.LOOKING); } } break; // 代表本机已经确认为Observer角色，正在集群内进行观察 case OBSERVING: // 选举流程中的Observer不起作用，因此这个流程暂不分析，等到 // 下一篇分析ZK集群的数据同步再来具体分析其作用 break; // 代表本机已经确认为Follower角色，正在跟随Leader case FOLLOWING: try { // 本机器的上一次轮询确定出了本机器为Follower角色 setFollower(makeFollower(logFactory)); // 开始执行Follower角色的工作：跟随Leader机器 follower.followLeader(); } finally { // 当本次Follower跟随的集群发生了异常时将会改变本机的 // 角色，重新设置成LOOKING状态选举出新Leader // 异常情况：1、Leader宕机，导致本集群不得不重新选举； // 2、本集群内其它的Follower宕机超过半数导致Leader // 投票数低于总数一半，进行重新选举。 // 如果是本机宕机程序直接死亡，不会进入到Finally块 follower.shutdown(); setFollower(null); setPeerState(ServerState.LOOKING); } break; // 代表本机已经确认为Leader角色，正在领导集群内的各个机器 case LEADING: try { // 本机器的上一次轮询确定出了本机器为Leader角色 setLeader(makeLeader(logFactory)); // 开始执行Leader角色的工作：作为集群中心发送同步命令 leader.lead(); // 退出了lead()方法说明集群的Leader发生了变化，需要 // 选举出新的Leader setLeader(null); }finally { // 关闭当前Leader对象并设置状态LOOKING开始准备下一次 // 选举流程 if (leader != null) { leader.shutdown("Forcing shutdown"); setLeader(null); } setPeerState(ServerState.LOOKING); } break; } } } finally { // 执行到这说明本机器的ZK服务被关闭，将会关闭机器的对象并退出 } }}3. FastLeaderElection选举发送通知
无论是刚刚启动或者是上一代的Leader退位开始选举新的Leader，各个机器在开始选举流程时的状态都是LOOKING，都会执行FastLeaderElection选举对象的公共流程。接下来便分析一下这个对象的公共流程，关键源码如下：
public class FastLeaderElection implements Election { // 本机器的集群对象 QuorumPeer self; // 选举流程时的逻辑迭代数，每调用一次lookForLeader进行选举时该值会+1 // 发送到其它机器上时对应Notification对象的electionEpoch属性 volatile long logicalclock; // 本机推崇将要当选leader的myid，对应Notification对象的leader，可以看成是 // 某个机器的id long proposedLeader; // 本机推崇将要当选leader的zxid，对应Notification对象的zxid long proposedZxid; // 本机推崇将要当选leader的epoch，对应Notification对象的peerEpoch long proposedEpoch; public Vote lookForLeader() throws InterruptedException { // 开始在集群内选举Leader，注册LeaderElection到JMX忽略 if (self.start_fle == 0) { // 记录FLE算法的开始时间 self.start_fle = System.currentTimeMillis(); } try { // 本集合key为leaderId，value为对应id的投票信息，集合将会记录 // 本次投票的各个机器投票情况 HashMap<Long, Vote> recvset = new HashMap<Long, Vote>(); // 新加入的机器用来记录集群内其它机器的投票情况 HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>(); // 每次轮询其它机器发来消息的间隔时间，固定200毫秒执行一次 int notTimeout = finalizeWait; synchronized(this){ // 逻辑选举次数+1，代表本机器有一次执行了重新选举Leader的操作 logicalclock++; // 投票前先把本机器的投票信息投给自己，getInitId()为本机器的 // myid值，getInitLastLoggedZxid()为本机器的zxid值 // getPeerEpoch()为本机器的currentEpoch值 updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch()); } // 对集群内的各个机器发送消息通知，告诉他们我选举自己当选Leader // 此时各个机器的通信对已经创建完毕，因此可以将消息发送给集群内的 // 各个机器，结果为A->B、A->C通知A当选Leader，B->A，B->C通知B当选 // Leader，C->B、C->A通知C当选Leader，例子中的三台机器每台机器都 // 会向集群内其它两台机器发送当选本机器为Leqader的消息通知，当然 // 也会通知自己，但是通知自己不会经过网络通信 sendNotifications(); // 发完通知消息后开始轮询其它机器的消息 while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){ // 轮询集合内是否有其它机器发来的消息，在本次三台机器的集群中， // recvqueue.poll()方法一定可以轮询出三个响应消息，其中一个 // 消息通知为本系统在前面的sendNotifications()方法发出的，没 // 经过网络通信，而是直接放在了本机的集合中等待处理 Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); // 后续收到通知处理流程这里暂不分析，等分析完本机器发送完通知后 // 再逐个分析 ... } } } synchronized void updateProposal(long leader, long zxid, long epoch){ // 更新本机器记录的Leader信息，投票前把这些信息改成本机器的，即先把票 // 投给自己 proposedLeader = leader; proposedZxid = zxid; proposedEpoch = epoch; } private void sendNotifications() { // 轮询配置文件中所配置的各个Server信息，并向每台机器发送通知 for (QuorumServer server : self.getVotingView().values()) { long sid = server.id; // 将本机器的信息封装，并发给myid为sid的机器 ToSend notmsg = new ToSend(ToSend.mType.notification, proposedLeader,// 第一次发送此值为本机器的myid proposedZxid,// 第一次发送此值为本机器的zxid logicalclock,// 第一次发送此值为本机器的logicalclock QuorumPeer.ServerState.LOOKING,// 本机器流程为LOOKING sid,// 目标机器的myid proposedEpoch);// 第一次发送此值为本机器的currentEpoch // 放入sendqueue集合中以便本选举对象的WorkerSender发送这些 // 通知消息给其它的机器 sendqueue.offer(notmsg); } }}4. WorkerSender选择机器并发送通知
前面已经说过了FLE对象将会把投票信息放入到sendqueue集合中，而这个集合便是FLE对象和WorkerSender对象的通信集合。接下来看下WorkerSender在拿到这些消息对象执行了什么操作：
class WorkerSender extends ZooKeeperThread { // 集群连接管理对象，WorkerSender实际上是该对象的内部类 QuorumCnxManager manager; // 将要使用通信对发送消息的消息存储队列集合，通信对发送消息时将会从该集合中 // 取出消息对象并使用Socket通信发送给对应的机器 LinkedBlockingQueue<ToSend> sendqueue; public void run() { // 启动通信对的发送信息对象后本方法将会被执行，直到该对象被调用finish() // 方法销毁 while (!stop) { try { // 从消息队列集合中获取需要发送的消息对象，固定阻塞3s，如果 // 没有轮询到则返回null ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS); // 如果为null说明暂时没有消息发送，继续轮回 if(m == null) continue; // 如果不为空则说明有需要发送的消息，调用process发送消息对象 process(m); } catch (InterruptedException e) { // 如果轮询集合发生异常则退出 break; } } } void process(ToSend m) { // 将需要发送的消息转换成Socket方便发送的ByteBuffer缓存对象 ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), m.leader, m.zxid, m.electionEpoch, m.peerEpoch); // 通知连接管理对象需要发送requestBuffer对象中的信息 manager.toSend(m.sid, requestBuffer); } static ByteBuffer buildMsg(int state, long leader, long zxid, long electionEpoch, long epoch) { // 生成ByteBuffer对象并封装byte[]数组，这里需要特别说明下各个参数和 // 在FLE中参数的对应关系 byte requestBytes[] = new byte[40]; ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes); requestBuffer.clear(); // 对应PeerQuorum中的peerState，此时值为LOOKING requestBuffer.putInt(state); // 对应PeerQuorum中的proposedLeader，刚开始选举为本机器的myid requestBuffer.putLong(leader); // 对应PeerQuorum中的proposedZxid，刚开始选举为本机器的zxid requestBuffer.putLong(zxid); // 对应PeerQuorum中的logicclock，代表本次选举的迭代数 requestBuffer.putLong(electionEpoch); // 对应PeerQuorum中的proposedEpoch，选举开始为本机器的currentEpoch requestBuffer.putLong(epoch); // 默认版本信息，接收到后会设置为接收消息的version属性 requestBuffer.putInt(Notification.CURRENTVERSION); return requestBuffer; }}5. QuorumCnxManager连接机器并发送消息
在将需要发送的消息转换成Socket发送消息的对象ByteBuffer后，现在面临一个问题：那便是本机器还尚未和其它机器简历Socket长连接通信，而QuorumCnxManager的职责便是管理连接，它会帮我们解决这个问题。关键源码如下：
public class QuorumCnxManager { // 每次发送消息的数量，固定是一，确保消息可以有序安全的发送出去 static final int SEND_CAPACITY = 1; // QuorumCnxManager对象和外界对象进行交互消息交互的集合中介，往这个集合中 // 放入数据说明一个问题：RecvWorker已经收到了其它机器的消息并处理转换完成 public final ArrayBlockingQueue<Message> recvQueue; // 接收集合recvQueue的容量 static final int RECV_CAPACITY = 100; // 将要发送给某个机器的ByteBuffer集合，key为发送机器的sid，value为单个消息 // 元素的阻塞队列，确保每次只发送一条消息（ArrayBlockingQueue长度固定） final Map<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap; // 保存和集群内另一台机器通信对的集合，key为另一台机器的myid，value则是 // 本机器与其通信的通信对 final ConcurrentHashMap<Long, SendWorker> senderWorkerMap; public void toSend(Long sid, ByteBuffer b) { // 如果要发送的myid等于本机器的id，不用发送，直接放入recvQueue集合中 // 需要注意的是recvQueue集合和前面在FLE对象中提到的recvqueue集合很像 // 这里做个简单说明：recvQueue集合是和FLE中的WorkerReceiver进行交互的 // recvqueue集合则是WorkerReceiver和真正的FLE对象交互的。交互对象需要 // 搞清楚，要不然看源码的时候很容易迷糊 if (self.getId() == sid) { b.position(0); // 直接添加到recvQueue集合中，相当于已经通过RecvWorker收到了消息 // 但是由于是发给自己的，因此忽略了RecvWorker这一步 addToRecvQueue(new Message(b.duplicate(), sid)); } else { // 如果集合中还没有sid的阻塞队列，则进行创建并放入到集合中 BlockingQueue<ByteBuffer> bq = queueSendMap.computeIfAbsent(sid, serverId -> new CircularBlockingQueue<>(SEND_CAPACITY)); // 再将需要发送的ByteBuffer对象消息放入到阻塞队列中 addToSendQueue(bq, b); // 真正开始根据sid去和对应的机器创建Socket长通信 connectOne(sid); } } public void addToRecvQueue(final Message msg) { // 将最新需要发送的消息添加到集合中 final boolean success = this.recvQueue.offer(msg); if (!success) { throw new RuntimeException("Could not insert into receive queue"); } } synchronized boolean connectOne(long sid, MultipleAddresses electionAddr) { if (senderWorkerMap.get(sid) != null) { if (self.isMultiAddressEnabled() && electionAddr.size() > 1 && self.isMultiAddressReachabilityCheckEnabled()) { senderWorkerMap.get(sid).asyncValidateIfSocketIsStillReachable(); } return true; } // 如果和一台机器的myid为sid没有创建过通信对则准备创建 return initiateConnectionAsync(electionAddr, sid); } public boolean initiateConnectionAsync(final MultipleAddresses electionAddr, final Long sid) { if (!inprogressConnections.add(sid)) { return true; } try { // todo connectionExecutor.execute(new QuorumConnectionReqThread(electionAddr, sid)); connectionThreadCnt.incrementAndGet(); } ... return true; } // 调用QuorumConnectionReqThread的run()方法 private class QuorumConnectionReqThread extends ZooKeeperThread { final MultipleAddresses electionAddr; final Long sid; QuorumConnectionReqThread(final MultipleAddresses electionAddr, final Long sid) { super("QuorumConnectionReqThread-" + sid); this.electionAddr = electionAddr; this.sid = sid; } @Override public void run() { try { initiateConnection(electionAddr, sid); } finally { inprogressConnections.remove(sid); } } } public void initiateConnection(final MultipleAddresses electionAddr, final Long sid) { // 创建Socket对象 Socket sock = null; try { LOG.debug("Opening channel to server {}", sid); if (self.isSslQuorum()) { sock = self.getX509Util().createSSLSocket(); } else { sock = SOCKET_FACTORY.get(); } // 在这个方法中设置timeout setSockOpts(sock); // 连接另外一台参与选举的机器，并且设置连接时间为5s sock.connect(electionAddr.getReachableOrOne(), cnxTO); ... } ... try { // todo startConnection(sock, sid); } ... } private boolean startConnection(Socket sock, Long sid) throws IOException { DataOutputStream dout = null; DataInputStream din = null; LOG.debug("startConnection (myId:{} --> sid:{})", self.getId(), sid); try { // Use BufferedOutputStream to reduce the number of IP packets. This is // important for x-DC scenarios. BufferedOutputStream buf = new BufferedOutputStream(sock.getOutputStream()); // 连接上另一台myid为sid的机器后立马向其发送本机器的myid dout = new DataOutputStream(buf); ... dout.write(addr_bytes); dout.flush(); din = new DataInputStream(new BufferedInputStream(sock.getInputStream())); } ... // authenticate learner QuorumPeer.QuorumServer qps = self.getVotingView().get(sid); if (qps != null) { // TODO - investigate why reconfig makes qps null. authLearner.authenticate(sock, qps.hostname); } // If lost the challenge, then drop the new connection // 这里是创建集群内通信结构的关键点之一，即在上一篇中写过的complete事件 // 在本次分析源码的假设三台机器A、B、C中，A的sid最小为1，B的sid居中为3 // C的sid最大为5，因此在各个机器中，A将会由于sid小于其它的机器而无法主动 // 建立通信对，B只能主动对A建立通信对，而C可以主动向B和A建立通信对。A和B // 的被动连接将会在后续分析Listener类中讲解。 // 简而言之，sid大的->sid小的=大的建立通信对，sid小的->sid大的=关闭连接 if (sid > self.getId()) { LOG.info("Have smaller server identifier, so dropping the connection: (myId:{} --> sid:{})", self.getId(), sid); // 当A->B、A->C和B->C这三种情况时会进入到这里面主动关闭本Socket // 解释：sid小的主动连接sid大的会主动关闭Socket连接。显示场景为： // B和C机器都已经启动了，而A是最后启动的，此时A机器执行到了这里， // A机器会主动的关闭连接 closeSocket(sock); // Otherwise proceed with the connection } else { LOG.debug("Have larger server identifier, so keeping the connection: (myId:{} --> sid:{})", self.getId(), sid); // 当C->A、C->B和B->A这三种情况时会进入到这里面主动创建通信对 // 解释：sid大的主动连接sid小的将会在本机器中主动创建通信对 // 根据传入的Socket对象创建通信对，需要注意的是通信对里面的sid是 // 需要进行通信的机器sid，而不是本机器的 // 现实场景为：A和B机器已经创建了，C最后启动的，此时C机器由于sid比 // A和B要大，因此会执行到这里，主动创建和A、B的通信对 SendWorker sw = new SendWorker(sock, sid); RecvWorker rw = new RecvWorker(sock, din, sid, sw); sw.setRecv(rw); // 获取以前选举通信时可能存在的通信对对象 SendWorker vsw = senderWorkerMap.get(sid); // 如果原来senderWorkerMap中有了sid对应的通信对，则拿出来主动销毁 // 因为通信对都是线程对象，可能存在以前选举时残留的数据，需要主动的 // 清空并关闭Socket连接，重新使用新的通信对对象 if (vsw != null) { vsw.finish(); } // 以sid为key，通信对为value放入到senderWorkerMap集合中 senderWorkerMap.put(sid, sw); // 如果消息发送集合中没有key为sid的阻塞队列则先创建放入集合中 // 再做一次确认，但在刚刚的流程中queueSendMap肯定已经被初始化并 // 放入了需要发送的数据的 queueSendMap.putIfAbsent(sid, new CircularBlockingQueue<>(SEND_CAPACITY)); // 启动发送消息线程对象，开始监听queueSendMap对象的阻塞队列 sw.start(); // 启动接收消息线程对象，用来接收对应机器发送来的消息 rw.start(); return true; } return false; } }经过了这个流程，在三台机器中的通信对情况如下图：
6 Listener监听Socket连接
在上面的四个流程中，经过在QuorumCnxManager对象建立连接后sid大的机器已经主动创建完了通信对，形成了上面图示的通信对情况，接下来要分析的是sid小的机器被动创建通信对的流程。关键源码如下：
public class Listener extends ZooKeeperThread { // 使用配置的electionPort端口+本机的地址创建的服务Socket，用来被动的和其它 // 机器进行交互（与其说被动的不如说给sid小的机器主动通信的机会） volatile ServerSocket ss = null; // 用负数来记录观察者的数量，并为其赋值负值来标明唯一性 private long observerCounter = -1; // 保存和集群内另一台机器通信对的集合，key为另一台机器的myid，value则是 // 本机器与其通信的通信对 final ConcurrentHashMap<Long, SendWorker> senderWorkerMap; // 将要发送给某个机器的ByteBuffer集合，key为发送机器的sid，value为单个消息 // 元素的阻塞队列，确保每次只发送一条消息（ArrayBlockingQueue长度固定） final Map<Long, ArrayBlockingQueue<ByteBuffer>> queueSendMap; @Override public void run() { int numRetries = 0; InetSocketAddress addr; // IO失败可重试三次 while((!shutdown) && (numRetries < 3)){ try { // 创建服务Socket对象 ss = new ServerSocket(); ss.setReuseAddress(true); // 获取本机器在配置中所配置的端口或者地址 if (self.getQuorumListenOnAllIPs()) { int port = self.quorumPeers.get(self.getId()). electionAddr.getPort(); addr = new InetSocketAddress(port); } else { addr = self.quorumPeers.get(self.getId()) .electionAddr; } // 设置本listener的地址名称 setName(self.quorumPeers.get(self.getId()).electionAddr .toString()); // 将服务Socket对象绑定该地址 ss.bind(addr); while (!shutdown) { // 开始接收其它机器发送过来的连接请求，sid大的或者sid小的 // 都会发送连接请求，在前面分析过，sid小的对sid大的机器发 // 送连接之后会主动关闭连接，其对sid大的机器创建通信对的操 // 作便是放在这个流程中 Socket client = ss.accept(); // 设置timeout时间为tickTime*syncLimit setSockOpts(client); // 接收到其它机器的请求后开始处理 receiveConnection(client); // 重试次数重置为0 numRetries = 0; } }// 异常捕获忽略 ... } } public void receiveConnection(Socket sock) { // 从名字也可以看出来这个方法就是用来接收Socket连接并处理的 // 和刚刚分析过的initiateConnection方法作用类似，只是 // initiateConnection方法是让sid大的主动创建通信对，而这个方法 // 则是让sid小的被动创建通信对 Long sid = null; try { // 在上面的initiateConnection方法中说了，在判断sid的大小值并处理 // 之前，连上Socket的第一件事便是把本机器的myid发送出去。举个例子： // A->C，由于A的sid比C的小，因此A不会主动创建和C的通信对，但连接 // 之后A会立马把自己的myid发送给C，而C->A时C也会主动的把自己的myid // 发送给A，从而各自触发Listener监听 DataInputStream din = new DataInputStream(sock .getInputStream()); // 读取其它机器发送过来的myid sid = din.readLong(); // 第一次接收的sid可能是version版本号 if (sid < 0) { // 如果是版本号则再次读取sid sid = din.readLong(); // 判断是否有剩余的数组需要读取 int num_remaining_bytes = din.readInt(); // 如果接下来的数据长度为负数或者大于了最大缓存值2048字节 // 则说明有问题，需要关闭连接（值如果是0也是OK的） if (num_remaining_bytes < 0 || num_remaining_bytes > maxBuffer) { closeSocket(sock); return; } // 将读取到的长度实例化一个数组并将剩余的读取完 byte[] b = new byte[num_remaining_bytes]; int num_read = din.read(b); } // 如果这个sid等于观察者的id，则将其赋值为observerCounter，每次 // 有新的观察者，observerCounter都会减一，保持sid的特殊性以及 // 观察者sid的唯一性 if (sid == QuorumPeer.OBSERVER_ID) { sid = observerCounter--; } } catch (IOException e) { closeSocket(sock); return; } // 看到这里又是熟悉的感觉，在initiateConnection方法中也有类似的场景 // 但是需要注意的是initiateConnection方法第一个if判断语句条件是 // “sid > self.getId()”，和本方法中的if判断相反，原因就是本方法实际上 // 就是initiateConnection方法的被动实现。 // 依然是A、B、C三台机器，我们已经确认了经过在initiateConnection方法中 // 执行完后的逻辑，C将会有B和A的通信对，而B将会有A的通信对，所有sid大的 // 机器都会有sid小的机器通信对，但是小的sid机器没有大的sid机器通信对。 // 以上述情况是根本无法做到集群内的机器互相通信的，因此需要本方法来补充 // 下面的逻辑大致为：sid小的可以在本机被动的创建和sid大的机器通信对；而 // sid大的机器接收到sid小的机器连接请求后，如果本机器没有sid小的机器的 // 通信对，则会关闭本次的Socket对象并在本机建立和sid小的机器的通信对。 if (sid < self.getId()) { // 进入到这里的情况是A->B、A->C、B->C，即sid小的机器向sid大的机器 // 发送请求连接，现实场景可以理解成A机器在C机器后面启动，A机器在启 // 动的时候向C机器发送连接请求，但由于C没启动，无法达到，因此作废。 // 而等到A机器启动时就会向C机器发送请求，此时C机器在监听到了A的请求 // 后便会遍执行到了这里 // 从本机器的senderWorkerMap集合取出可能存在的通信对 SendWorker sw = senderWorkerMap.get(sid); // 如果原来存在的将原来的通信对销毁释放 if (sw != null) { // 销毁通信对 sw.finish(); } // 关闭A或者B机器（即sid小的机器）的连接请求Socket对象 closeSocket(sock); // 调用已经分析过的connectOne方法，开始在本机器上再次主动创建 // 和A、B机器的通信对（即sid小的机器） connectOne(sid); } else { // 进入到这里的情况是C->A、C->B、B->A，即sid大的机器向sid小的机器 // 发送连接请求，此时sid小的机器监听后将会执行到这里开始在本机器中 // 被动的创建和sid大的机器的通信对 // 显示场景为：A和B机器已经启动了，但是C机器最后启动的，此时C机器 // 会向A和B机器发送连接请求，A和B机器由于sid小于C机器，因此监听到 // 连接请求后会执行到这里被动的创建和C机器的通信对 // 使用sid大的机器信息和Socket通信对象创建通信对 SendWorker sw = new SendWorker(sock, sid); RecvWorker rw = new RecvWorker(sock, sid, sw); sw.setRecv(rw); // 如果本机器原来有sid对应机器的通信对则销毁 SendWorker vsw = senderWorkerMap.get(sid); if(vsw != null) { // 调用销毁方法 vsw.finish(); } // 将新的通信对放入到senderWorkerMap集合中以便通信对可以监听 // 集合的消息变化 senderWorkerMap.put(sid, sw); // 如果保存要发送消息集合不包含新请求进来的sid对应机器则创建 if (!queueSendMap.containsKey(sid)) { queueSendMap.put(sid, new ArrayBlockingQueue<ByteBuffer>( SEND_CAPACITY)); } // 启动通信对发送消息线程对象，开始监听queueSendMap集合发送消息 sw.start(); // 启动通信对接收消息线程对象，开始监听其它机器的Socket消息并接收 rw.start(); return; } }}至此，集群内的各个机器通信对建立情况如下图：
经过这些流程，三台机器已经建立了和集群内每台机器的通信对，已经可以互相发送接收选举消息了，接下来便开始分析选举流程。
参考文章
zookeeper3.7版本github源码注释分析## zk源码分析系列Zookeeper原理和源码学习系列\ Zookeeper学习系列\ Zookeeper源码系列
原文：https://juejin.cn/post/7100494637638352910