手把手带你撸zookeeper源码-zookeeper启动(四)leader选举投票发送以及响应

秒速五厘米 2023-03-04 06:51 16阅读 0赞

上篇文章[手把手带你撸zookeeper源码-zookeeper启动(三)(zookeeper选举时是如何基于socket建立连接的)][zookeeper_-zookeeper_zookeeper_socket]写了zookeeper集群之间如何基于socket建立的连接，连接建立完毕之后，接下来就是开始发送投票和接收投票，然后对投票结果进行整理，最终选定哪台服务器是leader，哪个是follower

本篇文章我们接着看一下投票数据是如何发送出去的，以及如何接收到其他zk服务器发送过来的投票数据

结合上篇文章所讲，我们知道，QuorumCnxManager.toSend()方法中，在连接创建之前，会把投票数据封装为一个ByteBuffer,然后把数据放入到sid对应的queue队列中，即queueSendMap中，存放了哪些sid中有哪些数据要发送

然后在startConnection方法中，有如下代码

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);
                queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY));
                
                sw.start();
                rw.start();

启动了两个线程，并且发送数据线程和接受数据线程进行了绑定

我们进入到SendWorker线程中的run方法，猜测此线程中应该会拿queueSendMap中的数据，然后通过sid获取到每个要发送的数据队列，取出数据然后发送数据

while (running && !shutdown && sock != null) {
    
                        ByteBuffer b = null;
                        try {
                            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;
                            }
    
                            if(b != null){
                                lastMessageSent.put(sid, b);
                                send(b);
                            }
                        } catch (InterruptedException e) {
                            LOG.warn("Interrupted while waiting for message on queue",
                                    e);
                        }

直接能看到如上的代码，其实和我们猜测的差不多，就是会拿我们之前放入队列里面的投票数据，然后发送出去，我们直接进入到最后的send()方法，看看

synchronized void send(ByteBuffer b) throws IOException {
                byte[] msgBytes = new byte[b.capacity()];
                try {
                    b.position(0);
                    b.get(msgBytes);
                } catch (BufferUnderflowException be) {
                    LOG.error("BufferUnderflowException ", be);
                    return;
                }
                dout.writeInt(b.capacity());
                dout.write(b.array());
                dout.flush();
            }

这块很简单，就是直接通过输出流把数据write出去即可

好了，整个发送zk集群前期创建连接，发送自己的投票数据出去我们也剖析完了，大家如果有问题可以在评论区来一波评论

接下来我们继续，就是我把投票数据发送出去了，但是我是怎么接收到其他zk发送过来的投票的呢？

回到上面的代码，其实它创建了两个线程，一个是发送数据的线程，一个是接受数据的线程，那么接受数据的线程是不是就是读取其他zk发送过来的数据的呢？我们进入RecvWorker.run方法中去

while (running && !shutdown && sock != null) {
                        /**
                         * Reads the first int to determine the length of the
                         * message
                         */
                        int length = din.readInt();
                        if (length <= 0 || length > PACKETMAXSIZE) {
                            throw new IOException(
                                    "Received packet with invalid packet: "
                                            + length);
                        }
                        /**
                         * Allocates a new ByteBuffer to receive the message
                         */
                        byte[] msgArray = new byte[length];
                        din.readFully(msgArray, 0, length);
                        ByteBuffer message = ByteBuffer.wrap(msgArray);
                        addToRecvQueue(new Message(message.duplicate(), sid));
                    }

我也只是粘贴重要代码，就是RecvWorker会不断的循环读取其他zk发送过来的数据，如果读取到了数据，把它封装为一个Message对象，然后通过addToRecvQueue方法把数据放入到recvQueue队列中，然后肯定有一个其他的线程，会从recvQueue里面读取数据，然后进行处理

其实这块的代码有点绕，需要前后的代码关联起来，我们之前在创建FastLeaderElection的时候会启动WorkerReceiver线程，忘记的小伙伴可以看看之前的代码，我们直接进入WorkerReceiver.run方法中看看是不是获取recvQueue队列里面的数据然后进行消费的？

response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);

一眼就能看出来就是从recvQueue里面取出数据，然后进行数据的处理的

接着如下代码，是判断当前发送投票数据是否是observer，如果是，则执行下面的代码，如果不是，则走else代码，因为observer是不参与leader选举的，这块我们也不再看了

if(!validVoter(response.sid)){
                                Vote current = self.getCurrentVote();
                                ToSend notmsg = new ToSend(ToSend.mType.notification,
                                        current.getId(),
                                        current.getZxid(),
                                        logicalclock.get(),
                                        self.getPeerState(),
                                        response.sid,
                                        current.getPeerEpoch());
    
                                sendqueue.offer(notmsg);
                            }

接下来回去读取ByteBuffer里面的数据，然后封装到对象Notification中，我们看看发送过来的数据，封装到ByteBuffer中的数据是什么？

在创建连接的时候，会封装一个ToSend对象

ToSend notmsg = new ToSend(ToSend.mType.notification,
                        proposedLeader,
                        proposedZxid,
                        logicalclock.get(), //投票周期号
                        QuorumPeer.ServerState.LOOKING,
                        sid,
                        proposedEpoch);

然后会把这个对象中的属性进行包装成一个ByteBuffer，发送出去

在FastLeaderElection\#buildMsg方法中是来构建整个发送的ByteBuffer，这个之前有说到，就不再详细描述了

static ByteBuffer buildMsg(int state,
                long leader,
                long zxid,
                long electionEpoch,
                long epoch) {
            byte requestBytes[] = new byte[40];
            ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
    
            /*
             * Building notification packet to send 
             */
    
            requestBuffer.clear();
            requestBuffer.putInt(state);
            requestBuffer.putLong(leader);
            requestBuffer.putLong(zxid);
            requestBuffer.putLong(electionEpoch);
            requestBuffer.putLong(epoch);
            requestBuffer.putInt(Notification.CURRENTVERSION);
            
            return requestBuffer;
        }

把ByteBuffer中的数据封装为一个Notification对象

Notification n = new Notification();
                                
                                // State of peer that sent this message
                                QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
                                switch (response.buffer.getInt()) {
                                case 0:
                                    ackstate = QuorumPeer.ServerState.LOOKING;//启动时肯定是LOOKING
                                    break;
                                case 1:
                                    ackstate = QuorumPeer.ServerState.FOLLOWING;
                                    break;
                                case 2:
                                    ackstate = QuorumPeer.ServerState.LEADING;
                                    break;
                                case 3:
                                    ackstate = QuorumPeer.ServerState.OBSERVING;
                                    break;
                                default:
                                    continue;
                                }
                                
                                n.leader = response.buffer.getLong();
                                n.zxid = response.buffer.getLong();
                                n.electionEpoch = response.buffer.getLong();
                                n.state = ackstate;
                                n.sid = response.sid;
                                if(!backCompatibility){
                                    n.peerEpoch = response.buffer.getLong();
                                } else {
                                    if(LOG.isInfoEnabled()){
                                        LOG.info("Backward compatibility mode, server id=" + n.sid);
                                    }
                                    n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
                                }
    
                                n.version = (response.buffer.remaining() >= 4) ? 
                                             response.buffer.getInt() : 0x0;

接下来因为当前状态肯定是LOOKING状态，所以会走如下分支代码，就是要把这个Notification对象放入到recvqueue队列中，

if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
                                    recvqueue.offer(n);
    
                                    
                                    if((ackstate == QuorumPeer.ServerState.LOOKING)
                                            && (n.electionEpoch < logicalclock.get())){
                                        Vote v = getVote();
                                        ToSend notmsg = new ToSend(ToSend.mType.notification,
                                                v.getId(),
                                                v.getZxid(),
                                                logicalclock.get(),
                                                self.getPeerState(),
                                                response.sid,
                                                v.getPeerEpoch());
                                        sendqueue.offer(notmsg);
                                    }
                                }

接下来该干什么了呢？猜想一下，我把我的投票发送出去了，第一次肯定是投自己的选票，发送出去之后，接收到别人的投票，此时会读取别人的投票，然后对投票结果进行整合，看看是否有leader产生

回到我们开始的地方FastLeaderElection\#lookForLeader，当开始准备leader选举时，会先把自己的投票通过sendNotification方法发送出去，接下来通过while循环读取其他zk发送过来的数据

while ((self.getPeerState() == ServerState.LOOKING) &&
                        (!stop)){
                    Notification n = recvqueue.poll(notTimeout,
                            TimeUnit.MILLISECONDS);
    }

从recvqueue中获取到上面讲解的Notification对象

如果没有从recvqueue中读取都数据时，先走如下分支

if(n == null){
                        if(manager.haveDelivered()){
                            sendNotifications();
                        } else {
                            manager.connectAll();
                        }
    
                        /*
                         * Exponential backoff
                         */
                        int tmpTimeOut = notTimeout*2;
                        notTimeout = (tmpTimeOut < maxNotificationInterval?
                                tmpTimeOut : maxNotificationInterval);
                        LOG.info("Notification time out: " + notTimeout);
                    }

然后判断是否已经投递，如果是，则进行下一轮投票，如果没有，则和其他的zk创建连接，manage.connectAll和我们之前讲解的connectOne中的逻辑是一样的

而如果在recvqueue已经有了数据

if(validVoter(n.sid) && validVoter(n.leader)) {
                        switch (n.state) {
                        case LOOKING:
                            // If notification > current, replace and send messages out
                            if (n.electionEpoch > logicalclock.get()) {
                                logicalclock.set(n.electionEpoch);
                                recvset.clear();
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                        getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                    updateProposal(n.leader, n.zxid, n.peerEpoch);
                                } else {
                                    updateProposal(getInitId(),
                                            getInitLastLoggedZxid(),
                                            getPeerEpoch());
                                }
                                sendNotifications();
                            } else if (n.electionEpoch < logicalclock.get()) {
                                
                                break;
                            } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                    proposedLeader, proposedZxid, proposedEpoch)) {
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                                sendNotifications();
                            }
    
                            //投票归档
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
    
                            if (termPredicate(recvset,
                                    new Vote(proposedLeader, proposedZxid,
                                            logicalclock.get(), proposedEpoch))) {
    
                                // Verify if there is any change in the proposed leader
                                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) {
                                    self.setPeerState((proposedLeader == self.getId()) ?
                                            ServerState.LEADING: learningState());
    
                                    Vote endVote = new Vote(proposedLeader,
                                                            proposedZxid,
                                                            logicalclock.get(),
                                                            proposedEpoch);
                                    leaveInstance(endVote);
                                    return endVote;
                                }
                            }
                            break;
                    }
    }

接下来肯定会走上面的分支，大家想想是不是？validVoter()主要是校验一下当前发送的zk是否有资格进行投票选举leader,如果有，则获取选票，如果没有，则会走else分支，然后打印一些日志，这里就不粘贴出来了。另外也会同时校验一下zk发送的投票数据选举的leader是否有资格被选举为leader，如果没有资格，也不会走这里

if (n.electionEpoch > logicalclock.get()) {
                                logicalclock.set(n.electionEpoch);
                                recvset.clear();
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                        getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                    updateProposal(n.leader, n.zxid, n.peerEpoch);
                                } else {
                                    updateProposal(getInitId(),
                                            getInitLastLoggedZxid(),
                                            getPeerEpoch());
                                }
                                sendNotifications();
                            } else if (n.electionEpoch < logicalclock.get()) {
                                
                                break;
                            } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
                                    proposedLeader, proposedZxid, proposedEpoch)) {
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                                sendNotifications();
                            }

这块的一大段代码，就是判断一系列的参数，然后来回进不断更新，不断投票的

protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
            LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" +
                    Long.toHexString(newZxid) + ", proposed zxid: 0x" + Long.toHexString(curZxid));
            if(self.getQuorumVerifier().getWeight(newId) == 0){
                return false;
            }
            
            /*
             * We return true if one of the following three cases hold:
             * 1- New epoch is higher    拥有一个更高的epoch
                //如果epoch相同，则zxid更高
             * 2- New epoch is the same as current epoch, but new zxid is higher
                // 如果epoch和zxid都一样，就看sid
             * 3- New epoch is the same as current epoch, new zxid is the same
             *  as current zxid, but server id is higher.
             */
            
            return ((newEpoch > curEpoch) || 
                    ((newEpoch == curEpoch) &&
                    ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
        }

我们看看上面的方法，它就是会对当前的投票数据进行排序，判断接下来要投递哪台服务器作为leader，然后调用sendNotification把投票再发出去

OK，到这里，我们基本上把zk的leader选举投票阶段的代码研读完了，接下里就是对投票进行归档，然后选择出来leader，接下来各个zk服务器判断自己是leader还是follower然后作为自己的角色启动就行了

我们可以大概划个流程图吧

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70][]

上面的图是整个发送数据的流程图，让大家更容易的去理解整个发送和接受数据的流程

下篇文章我们要开始梳理整个投票的归档，然后选举出一个leader，然后zk服务器可以根据自己的角色启动，然后做自己的事情了

[zookeeper_-zookeeper_zookeeper_socket]: https://blog.csdn.net/liereli/article/details/107732477
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70]: /images/20230208/c92a44ab6bbd47148a28a098878fb36d.png