手把手带你撸zookeeper源码-zookeeper启动(三)(zookeeper选举时是如何基于socket建立连接的)

悠悠 2023-03-04 06:35 102阅读 0赞

接上文 [手把手带你撸zookeeper源码-zookeeper启动(二)][zookeeper_-zookeeper]

结合上两篇文章写的zookeeper启动时，都干了些什么，画了一张图，把涉及到的组件都在图中给展示出来，大家可以参考一下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70][]

在这个图片中把之前写的源码中的所有组件都给包括了，以及每个组件是做什么用的，接下来我们会一步步的去梳理整个zookeeper源码，也会把这张图给进一步完善，最终形成一个完整的zookeeper架构原理图的

上篇文章最后，我们知道调用到QuorumPeer.start方法中，在这个方法中做了好几件事，我们在上篇文章中都已经讲过了，这里不再描述

@Override
        public synchronized void start() {
            //加载快照文件数据到内存中恢复数据
            loadDataBase();
            cnxnFactory.start();
            //启动leader选举
            startLeaderElection();
            //initLeaderElection() 为leader选举做好初始化工作
            super.start();
        }

大家可以看着上面的代码回顾一下之前讲的内容

最后调用了super.start()启动了一个线程, 它最终会调用QuorumPeer中的run方法，这个方法也是最终完成leader选举的方法，然后根据选举结果，判断出自己是Leader还是Follower或者是Observer，然后自己作为相应的角色，去做一系列属于本角色的事情

先把代码贴出来，逐步讲解

while (running) {
                    switch (getPeerState()) {//默认LOOKING
                    case LOOKING://开始选举leader
                        LOG.info("LOOKING");
    
                        if (Boolean.getBoolean("readonlymode.enabled")) {
                            LOG.info("Attempting to start ReadOnlyZooKeeperServer");
    
                            // Create read-only server but don't start it immediately
                            final ReadOnlyZooKeeperServer roZk = new ReadOnlyZooKeeperServer(
                                    logFactory, this,
                                    new ZooKeeperServer.BasicDataTreeBuilder(),
                                    this.zkDb);
        
                            Thread roZkMgr = new Thread() {
                                public void run() {
                                    try {
                                        // lower-bound grace period to 2 secs
                                        sleep(Math.max(2000, tickTime));
                                        if (ServerState.LOOKING.equals(getPeerState())) {
                                            roZk.startup();
                                        }
                                    } catch (InterruptedException e) {
                                        LOG.info("Interrupted while attempting to start ReadOnlyZooKeeperServer, not started");
                                    } catch (Exception e) {
                                        LOG.error("FAILED to start ReadOnlyZooKeeperServer", e);
                                    }
                                }
                            };
                            try {
                                roZkMgr.start();
                                setBCVote(null);
                                // lookForLeader开始投票选择leader
                                setCurrentVote(makeLEStrategy().lookForLeader());
                            } catch (Exception e) {
                                LOG.warn("Unexpected exception",e);
                                setPeerState(ServerState.LOOKING);
                            } finally {
                                // If the thread is in the the grace period, interrupt
                                // to come out of waiting.
                                roZkMgr.interrupt();
                                roZk.shutdown();
                            }
                        } else {
                            try {
                                setBCVote(null);
                                setCurrentVote(makeLEStrategy().lookForLeader());
                            } catch (Exception e) {
                                LOG.warn("Unexpected exception", e);
                                setPeerState(ServerState.LOOKING);
                            }
                        }
                        break;
                    case OBSERVING://当前zk角色为Observer
                        try {
                            LOG.info("OBSERVING");
                            setObserver(makeObserver(logFactory));
                            observer.observeLeader();
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception",e );                        
                        } finally {
                            observer.shutdown();
                            setObserver(null);
                            setPeerState(ServerState.LOOKING);
                        }
                        break;
                    case FOLLOWING://当前zk角色为Follower
                        try {
                            LOG.info("FOLLOWING");
                            setFollower(makeFollower(logFactory));
                            follower.followLeader();
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception",e);
                        } finally {
                            follower.shutdown();
                            setFollower(null);
                            setPeerState(ServerState.LOOKING);
                        }
                        break;
                    case LEADING://当前zk角色为Leader
                        LOG.info("LEADING");
                        try {
                            setLeader(makeLeader(logFactory));
                            leader.lead();
                            setLeader(null);
                        } catch (Exception e) {
                            LOG.warn("Unexpected exception",e);
                        } finally {
                            if (leader != null) {
                                leader.shutdown("Forcing shutdown");
                                setLeader(null);
                            }
                            setPeerState(ServerState.LOOKING);
                        }
                        break;
                    }
                }

把主要代码给贴出来，其他的无关紧要，可以不看

当我们刚启动一个zk时，它肯定回去寻找集群中的leader，如果集群中已经有leader了，那么会直接把自己作为一个follower角色启动。如果集群中还没有leader，则会先发起投票。首先每个zk发起投票时的第一票都会先投自己的选票，然后接收其他zk发送过来的投票，每轮投票结束后，zk都会整理自己的投票结果，然后根据投票结果判断是否要发起第二轮选票，如果发起第二轮选票，zk会根据当前自己收到的投票结果中，根据三个规则来进行投票(1、首先会现根据leader的epoch版本来判断，epoch越大，越优先选为leader, 2、首先判断zxid最大的，越被优先为leader，因为zxid越大代表和leader数据同步的越相近，越小说明和leader数据同步差距越大。3、如果zxid相同，则根据myid来判断，如果myid越大，则越优先被选为leader)

分析上面的代码，判断当前是否设置了readonlymode.enabled = true，如果没有设置则会走else,如下

setCurrentVote(makeLEStrategy().lookForLeader());//选举leader

这里交给大家一个看源码技巧，之前一直强调抓大放下，如果你不确定它是走if，还是走else，这个可以不用管它，你看源码中，不管走哪个分支，都会走上面两行代码，并且这行代码一看就是选举leader，核心重中之重，对不对？不要纠结于细节

这里面包含了三个方法，我们来逐步分析一下

makeLEStrategy()

这方法很简单，就是直接返回一个Election选举实例，它是一个借口，它的实现类就是我们之前一直在说的FastLeaderElection

说明调用lookForLeader方法时，会直接进入到FastLeaderElection.lookForLeader()方法中

在这个里面就是要选举leader做的一系列动作，我们进去看看都做了什么事。进去之后发现一大坨代码，一眼看去不知道做了什么事，我也不把代码一下子全部粘贴出来了，我会从上往下，把要讲的内容，逐一粘贴，这样理解起来更容易些。当我们看到一个方法中有一大坨代码时，不要着急心慌，抓耳挠腮的，蒙蒙的不知道该咋看，就一行一行地看，连蒙带猜是否和我们的主线有关，如果有关再进去看，如果没关系，就直接略过

刚开始的几行代码一看就是和jmx监控有关系的，我们不看，然后接着往下看

HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
    HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
    int notTimeout = finalizeWait;
    synchronized(this){
        logicalclock.incrementAndGet();
        updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
    }
    LOG.info("New election. My id =  " + self.getId() +
                        ", proposed zxid=0x" + Long.toHexString(proposedZxid));
    sendNotifications();//发起自己的投票给别人, 把投票放入sendQueue

我们先看前面的几行代码，看看都做了什么

一眼定位到synchronized这个代码块中，我们先分析一下它是干啥的

synchronized(this){
        logicalclock.incrementAndGet();
        updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
    }

logicalclock就是一个自增的原子AtomicLong，大概意思就是每次接收到其他的zk发送过来一次投票就会进行一次累加

synchronized void updateProposal(long leader, long zxid, long epoch){
            if(LOG.isDebugEnabled()){
                LOG.debug("Updating proposal: " + leader + " (newleader), 0x"
                        + Long.toHexString(zxid) + " (newzxid), " + proposedLeader
                        + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
            }
            proposedLeader = leader;
            proposedZxid = zxid;
            proposedEpoch = epoch;
        }

updateProposal(), 我们看打印的日志或者是注释，也是一种帮助我们去理解代码的好方式，其实就是接收到其他zk发送过来的leader(就是发送过来的投票选择哪个zk为leader，就是投票给那个zk的myid)、zxid、epoch，然后更新一下本地的的这几个变量, 最后会把这几个变量封装为Vote投票对象，作为一个投票

今天要讲的关键点就是sendNotifition()方法，这个方法里面封装了自己的投票信息，然后把它发送出去

private void sendNotifications() {
            for (QuorumServer server : self.getVotingView().values()) {
                long sid = server.id;
                //对其他机器都发起投票请求
                ToSend notmsg = new ToSend(ToSend.mType.notification,
                        proposedLeader,
                        proposedZxid,
                        logicalclock.get(), //投票周期号
                        QuorumPeer.ServerState.LOOKING,
                        sid,
                        proposedEpoch);
                if(LOG.isDebugEnabled()){
                    LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x"  +
                          Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock.get())  +
                          " (n.round), " + sid + " (recipient), " + self.getId() +
                          " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
                }
                sendqueue.offer(notmsg);
            }
        }

最后会轮询所有的其他zk server，然后把投票信息封装为一个ToSend，然后放入到sendqueue队列里面，发送出去。看到这有没有和我们前面讲的地方联系起来？好好想想

上篇文章中，我们有讲到在创建一个FastLeaderElection对象的时候，会调用一个starter方法，同时会创建两个队列，一个是sendQueue，用于存放待发送的数据，一个recvQueue，用来存放响应的通知数据，还有一个messenger，这个对量里面会启动两个线程，一个是WorkerSender，发送数据线程，和WorkerReceiver线程，用户接受数据的线程，我们看一下代码

private void starter(QuorumPeer self, QuorumCnxManager manager) {
            this.self = self;
            proposedLeader = -1;
            proposedZxid = -1;
            //创建一个发送数据队列
            sendqueue = new LinkedBlockingQueue<ToSend>();
            // 接受响应通知 队列
            recvqueue = new LinkedBlockingQueue<Notification>();
            //实例化一个messager
            this.messenger = new Messenger(manager);
        }

然后进入Messenger构造方法中

Messenger(QuorumCnxManager manager) {
                //创建一个发送数据线程
                this.ws = new WorkerSender(manager);
    
                Thread t = new Thread(this.ws,
                        "WorkerSender[myid=" + self.getId() + "]");
                t.setDaemon(true);
                t.start();
    
                //接受数据线程
                this.wr = new WorkerReceiver(manager);
                t = new Thread(this.wr,
                        "WorkerReceiver[myid=" + self.getId() + "]");
                t.setDaemon(true);
                t.start();
            }

创建完线程之后，就立即以保护模式启动了线程，我们可以进入WorkerSender的run方法中

public void run() {
                    while (!stop) {
                        try {
                            ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
                            if(m == null) continue;
    
                            process(m);
                        } catch (InterruptedException e) {
                            break;
                        }
                    }
                    LOG.info("WorkerSender is down");
                }

这块在上一篇文章中都有讲到，当做回顾再深入了解一下，WorkerSender会不停的从sendQueue里面取数据，然后如果有数据则会调用process方法发送出去

前后是不是关联起来了？一个把ToSend要发送的数据放入到sendqueue队列里面，一个线程在不断轮训拿队列里面的数据，一旦有数据放入队列，这边就会立即读取到

下面我们进入到process方法中，看看怎么真正的把ToSend真正发送出去的？

void process(ToSend m) {
         ByteBuffer requestBuffer = buildMsg(m.state.ordinal(), 
                                                            m.leader,
                                                            m.zxid, 
                                                            m.electionEpoch, 
                                                            m.peerEpoch);
         manager.toSend(m.sid, requestBuffer);
    }

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缓冲区，然后把要发送的数据放入缓冲区，调用manager.toSend给发送出去

public void toSend(Long sid, ByteBuffer b) {
            /*
             * If sending message to myself, then simply enqueue it (loopback).
             */
            if (this.mySid == sid) {
                 b.position(0);
                 addToRecvQueue(new Message(b.duplicate(), sid));
            } else {
                 /*
                  * Start a new connection if doesn't have one already.
                  */
                 ArrayBlockingQueue<ByteBuffer> bq = new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY);
                 ArrayBlockingQueue<ByteBuffer> bqExisting = queueSendMap.putIfAbsent(sid, bq);
                 if (bqExisting != null) {
                     addToSendQueue(bqExisting, b);
                 } else {
                     addToSendQueue(bq, b);
                 }
                 connectOne(sid);
                    
            }
        }

看一下注释，第一个就是先判断数据是否是发送给自己，如果发送给自己则怎么去处理，这块不讲

我们主要看else部分的代码，先看注释，很容易理解，就是是否和这个要发送sid对应的zk已经建立了连接，如果没有则新建一个

建立连接之前，会先把当前要发送的ByteBuffer放到queueSendMap里面对应的sid的队列里面，其实大家在这块应该能猜想到一点就是，queueSendMap里面有每个sid对应的一个队列，然后你要往哪个sid发送数据，就先把数据放入到sid对应的队列里面，然后应该会有其他的线程会把数据取出来发送出去，我们只要记住在这会往sid对应的队列面放数据即可，详细的我们稍后再看，现在看关键点就是connectionOne()这个方法，这个方法里面就是和其他zk建立连接

synchronized public void connectOne(long sid){
            if (!connectedToPeer(sid)){
                InetSocketAddress electionAddr;
                if (view.containsKey(sid)) {
                    electionAddr = view.get(sid).electionAddr;
                } else {
                    LOG.warn("Invalid server id: " + sid);
                    return;
                }
                try {
    
                    LOG.debug("Opening channel to server " + sid);
                    // 直接基于java socket进行tcp协议通信的, nio
                    Socket sock = new Socket();
                    setSockOpts(sock);
                    sock.connect(view.get(sid).electionAddr, cnxTO);
                    LOG.debug("Connected to server " + sid);
                    if (quorumSaslAuthEnabled) {
                        initiateConnectionAsync(sock, sid);
                    } else {
                        initiateConnection(sock, sid);
                    }
                } catch (UnresolvedAddressException e) {
                    if (view.containsKey(sid)) {
                        view.get(sid).recreateSocketAddresses();
                    }
                    throw e;
                } catch (IOException e) {
                    if (view.containsKey(sid)) {
                        view.get(sid).recreateSocketAddresses();
                    }
                }
            } else {
                LOG.debug("There is a connection already for server " + sid);
            }
        }

我把一部分注释干掉了，看着更清爽一些，自己写过基于socket编程的小伙伴一下子就能看出来，代码是干嘛的，先创建了一个Socket，然后调用sock.connection去建立连接

在这我们我看一下setSockOpts，设置了立即发送数据出去，和tcpKeepAlive,以及超时时间

sock.setTcpNoDelay(true);// 立即发送出去
    sock.setKeepAlive(tcpKeepAlive);
    sock.setSoTimeout(socketTimeout);

我们接着看initiateConnection-->startConnection

最后回到startConnection方法中，启动一个连接，我们直接看这个方法里面的代码

private boolean startConnection(Socket sock, Long sid)
                throws IOException {
            DataOutputStream dout = null;
            DataInputStream din = null;
            try {
                // Sending id and challenge
                dout = new DataOutputStream(sock.getOutputStream());
                dout.writeLong(this.mySid);
                dout.flush();
    
                din = new DataInputStream(
                        new BufferedInputStream(sock.getInputStream()));
            } catch (IOException e) {
                LOG.warn("Ignoring exception reading or writing challenge: ", e);
                closeSocket(sock);
                return false;
            }
    
            // authenticate learner
            authLearner.authenticate(sock, view.get(sid).hostname);
    
            // If lost the challenge, then drop the new connection
            //不允许和比自己sid大的机器建立连接
            if (sid > this.mySid) {
                LOG.info("Have smaller server identifier, so dropping the " +
                         "connection: (" + sid + ", " + this.mySid + ")");
                closeSocket(sock);
                // Otherwise proceed with the connection
            } 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);
                queueSendMap.putIfAbsent(sid, new ArrayBlockingQueue<ByteBuffer>(SEND_CAPACITY));
                
                sw.start();
                rw.start();
                
                return true;    
                
            }
            return false;
        }

有没有似曾相识的感觉？

在这里就完成了zk之间的连接建立，然后就可以发送数据了

这里剧透一下接下来要讲的内容，就是刚才封装的ToSend数据封装成的ByteBuffer是什么时候发送出去的呢？这个里面封装了自己的投票信息呢？我们下篇文章再讲

[zookeeper_-zookeeper]: https://blog.csdn.net/liereli/article/details/107676650
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZXJlbGk_size_16_color_FFFFFF_t_70]: /images/20230208/bcac74a2b96e4908a692ec2fbf799176.png