手把手带你撸zookeeper源码-从源码角度分析zookeeper启动时都做了什么？

╰半橙微兮° 2023-03-02 05:14 3阅读 0赞

# 前言 #

上篇文章 [手把手带你撸zookeeper源码(如何启动一个zookeeper服务)][zookeeper_zookeeper] 主要是讲解了如何通过脚本启动一个zk服务，还有一些简单的基本操作，以及如何去根据我们的服务器配置来设置JVM参数，接下来的文章就开始干zk源码了，这篇文章我们先找找启动zookeeper服务的入口，然后看看zookeeper启动的时候都做写什么？

# 看源码技巧 #

相信各位小伙伴大家都应该看过源码，但很多时候看源码要么就是一掠而过，方法调用方法，进去出来，根本就没有想过为什么这么设计？有没有用什么设计模式？有没有什么可以借鉴的设计思想？或者有没有问题？有没有优化的点？要么就是看代码过于细节，一行行的看，一个方法一个方法点击去看，然后就绕晕到各种方法调用里面，出不来，不知道都干了什么。我给大家八个字建议: 看主流程、抓大放小

看主流程:  你可以把源码当成一棵树，树有主干，有次主干，有枝叶，我们看源码的时候，先看主干，在看主干代码的过程当中肯定会有许多小分支(次主干)，那么第一次看的时候直接略过，不要纠结，也不要妄想每行代码都看懂，那是不可能的。除非是你自己开发的代码，否则一个开源项目你能弄懂它核心代码就足够了，一般一个开源项目的核心代码也就占整个项目的40%左右。所以一定要看主流程，其他的放弃。针对一个方法，不同的流程，你可能来回看好多遍，只是每次的重点不一样

抓大放小: 比如在代码中封装了针对数组的排序，以及对某个字符串做截取处理等等，这样的代码大概知道什么意思就行，没必要细看，主要就是看和主流程相关的方法代码。

一定一定不要纠结于细节

# 找入口 #

回想一下上节课我们怎么启动一个zk服务的？在bin目录下有一个zkServer.sh脚本，我们执行./zkServer.sh start来启动一个服务，那么我们就应该从这个脚本入手，看看在执行上面脚本的时候都执行了什么？

![20200727221832768.png][]

找到脚本中start参数对应的执行命令，我们很快就发现了nohup $JAVA, 其实说白了，zk的源码会被打成jar包，启动zk服务的时候就是利用java -jar来执行jar包中的main方法，然后启动一个服务的

$ZOOMAIN 就是main方法所在的类

$ZOOCFG 对应的是zoo.cfg

$CLASSPATH 类路径classpath

$JVMFLAGS  JVM配置的一些参数

![20200727222257906.png][]

找到ZOOMAIN时，是不是感觉忽然发现了宝藏？没错，QuorumPeerMain就是入口处，这个类里面肯定有main方法

当然我们也可以在启动完zk之后，通过jps来找到对应的入口，如下

![20200726062944835.png][]

# 源码分析 #

我们在看看QuorumPeerMain里面的main方法之前，先根据类名来猜测一下这个类是干啥的？

Quorum: 大多数        Peer to Peer 点对点，代表了一个zk服务进程，Quorum这个和zk的选举、2PC节点相关，大多数或者过半机制

看看main方法，我也都写上了注释

![20200727222853880.png][]

接下来 initializeAndRun方法中有几个关键的点，我们来一一分析一下

protected void initializeAndRun(String[] args)
            throws ConfigException, IOException
        {
            //用于解析配置文件的类
            QuorumPeerConfig config = new QuorumPeerConfig();
            if (args.length == 1) {
                //如果传入了一个参数，则认为是配置文件 zoo.cfg
                config.parse(args[0]);
            }
    
            // Start and schedule the the purge task
            // 定期清理日志文件和快照文件
            DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
                    .getDataDir(), config.getDataLogDir(), config
                    .getSnapRetainCount(), config.getPurgeInterval());
            purgeMgr.start();
    
            if (args.length == 1 && config.servers.size() > 0) {
                //集群模式
                runFromConfig(config);
            } else {
                LOG.warn("Either no config or no quorum defined in config, running "
                        + " in standalone mode");
                // there is only server in the quorum -- run as standalone
                //单机模式
                ZooKeeperServerMain.main(args);
            }
        }

第一个就是 config.parse()方法，其实简单点就是解析一下zoo.cfg，然后把解析出来的配置信息封装到了QuorumPeerConfig这个类的属性里面，类似于这样的代码，我们大概知道是什么意思就行了，没必要过于纠结于细节是如何解析，如何赋值到QuorumPeerConfig的属性里面的，这就是抓大放小，不要钻牛角尖，钻的太深

第二个关键点

DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config
                    .getDataDir(), config.getDataLogDir(), config
                    .getSnapRetainCount(), config.getPurgeInterval());
            purgeMgr.start();

我们根据类名来猜测一下，就是数据目录清理管理组件，大概就是清理一下文件之类操作，然后启动了一个线程任务去执行。我们现在不知道它主要来清除哪些文件数据的，那么就先把它放在这，有用到的回过头来再细看。(其实就是启动一个定时任务，然后去清理我们的事务日志文件和快照文件的)，到后面的时候我们会回过头来再细讲

第三个关键点

if (args.length == 1 && config.servers.size() > 0) {
                //集群模式
                runFromConfig(config);
            } else {
                LOG.warn("Either no config or no quorum defined in config, running "
                        + " in standalone mode");
                // there is only server in the quorum -- run as standalone
                //单机模式
                ZooKeeperServerMain.main(args);
            }

config.servers这个就是我们从zoo.cfg里面配置的server.1、server.2等信息，如果我们没有配置，说明是启动一个单机版的，如果配置了，则启动的是一个集群。我们这个主要是看集群是如何启动的，以及以后剖析如何建立连接，如何进行选举的

public void runFromConfig(QuorumPeerConfig config) throws IOException {
          
              ServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory();
              cnxnFactory.configure(config.getClientPortAddress(),
                                    config.getMaxClientCnxns());
    
              //一个peer相当于一个zk节点
              quorumPeer = getQuorumPeer();
    
              quorumPeer.setQuorumPeers(config.getServers());
              quorumPeer.setTxnFactory(new FileTxnSnapLog(
                      new File(config.getDataLogDir()),
                      new File(config.getDataDir())));
              quorumPeer.setElectionType(config.getElectionAlg());//默认为3
              quorumPeer.setMyid(config.getServerId());
              quorumPeer.setTickTime(config.getTickTime());
              quorumPeer.setInitLimit(config.getInitLimit());
              quorumPeer.setSyncLimit(config.getSyncLimit());
              quorumPeer.setQuorumListenOnAllIPs(config.getQuorumListenOnAllIPs());
              quorumPeer.setCnxnFactory(cnxnFactory);
              quorumPeer.setQuorumVerifier(config.getQuorumVerifier());
              quorumPeer.setClientPortAddress(config.getClientPortAddress());
              quorumPeer.setMinSessionTimeout(config.getMinSessionTimeout());
              quorumPeer.setMaxSessionTimeout(config.getMaxSessionTimeout());
              quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));
              quorumPeer.setLearnerType(config.getPeerType());
              quorumPeer.setSyncEnabled(config.getSyncEnabled());
    
              // sets quorum sasl authentication configurations
              quorumPeer.setQuorumSaslEnabled(config.quorumEnableSasl);
              if(quorumPeer.isQuorumSaslAuthEnabled()){
                  quorumPeer.setQuorumServerSaslRequired(config.quorumServerRequireSasl);
                  quorumPeer.setQuorumLearnerSaslRequired(config.quorumLearnerRequireSasl);
                  quorumPeer.setQuorumServicePrincipal(config.quorumServicePrincipal);
                  quorumPeer.setQuorumServerLoginContext(config.quorumServerLoginContext);
                  quorumPeer.setQuorumLearnerLoginContext(config.quorumLearnerLoginContext);
              }
    
              quorumPeer.setQuorumCnxnThreadsSize(config.quorumCnxnThreadsSize);
              quorumPeer.initialize();
    
              quorumPeer.start();
              quorumPeer.join();
         
        }

我对代码做了一些删减，都是些无关紧要的代码，我们看一下上面的代码，首先利用了一个工厂方法来创建了ServerCnxnFactory，顾名思义: server 连接工厂，应该是和zk集群服务之间的连接有关

static public ServerCnxnFactory createFactory() throws IOException {
            String serverCnxnFactoryName =
                System.getProperty(ZOOKEEPER_SERVER_CNXN_FACTORY);
            if (serverCnxnFactoryName == null) {
                serverCnxnFactoryName = NIOServerCnxnFactory.class.getName();
            }
                ServerCnxnFactory serverCnxnFactory = (ServerCnxnFactory) Class.forName(serverCnxnFactoryName)
                        .getDeclaredConstructor().newInstance();
                LOG.info("Using {} as server connection factory", serverCnxnFactoryName);
                return serverCnxnFactory;
            
        }

如果环境变量中没有zookeeper.serverCnxnFactory我们自己配置的factory，则创建默认的NIOServerCnxnFactory，通过反射来进行实例化，这个类是一个很重要的组件

接下来的代码就是创建了一个QuorumPeer对象，然后把从zoo.cfg中解析的属性信息从QuorumPeerConfig给QuorumPeer设置进去

这个关注一下

quorumPeer.setElectionType(config.getElectionAlg());

如果用户没有设置选举类型，则默认electionType  = 3，这个是用来指定选举算法的，后面会讲到

最后会调用quorumPeer.start()

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

synchronized public void startLeaderElection() {
        	
        	//自己一开始的选票 myid zxid  currentEpoch投票版本号
        	currentVote = new Vote(myid, getLastLoggedZxid(), getCurrentEpoch());
        	
            for (QuorumServer p : getView().values()) {
                if (p.id == myid) {
                    myQuorumAddr = p.addr;
                    break;
                }
            }
            // 0 基于UDP, electionType默认3
            if (electionType == 0) {
                try {
                    udpSocket = new DatagramSocket(myQuorumAddr.getPort());
                    responder = new ResponderThread();
                    responder.start();
                } catch (SocketException e) {
                    throw new RuntimeException(e);
                }
            }
            this.electionAlg = createElectionAlgorithm(electionType);
        }

注释，上面的几件事

1、loadDataBase(): 加载快照文件中的数据到内存中，主要是重启、故障恢复时快速恢复数据的，先不看

2、cnxnFactory.start(): 启动一个线程，zk集群之间同步数据通信使用的，以后看

3、startLeaderElection()启动选举，我们可以先看看使用了什么选举算法

Vote 投票，封装了三个参数，1、myid自己配置的，2、最新的zxid， 3、epoch可以当初leader的版本，每选举一次leader加一

看一下最后的createElectionAlgorithm方法，这个方法就是创建leader选举算法 FastLeaderElection

protected Election createElectionAlgorithm(int electionAlgorithm){
            Election le=null;
                    
            //TODO: use a factory rather than a switch
            //0 1 2 已经废弃
            //electionAlgorithm = 3
            switch (electionAlgorithm) {
            case 0:
                le = new LeaderElection(this);
                break;
            case 1:
                le = new AuthFastLeaderElection(this);
                break;
            case 2:
                le = new AuthFastLeaderElection(this, true);
                break;
            case 3:
                //zk节点网络通信的组件
                qcm = createCnxnManager();
                QuorumCnxManager.Listener listener = qcm.listener;
                if(listener != null){
                    // 启动一个listener监听，用于监听其他机器发送过来的请求
                    listener.start();
                    le = new FastLeaderElection(this, qcm);
                } else {
                    LOG.error("Null listener when initializing cnx manager");
                }
                break;
            default:
                assert false;
            }
            return le;
        }

最后因为QuorumPeer本身就是一个线程，super.start()就是启动了自己这个线程，执行QuorumPeer中的run方法，接下来就是选举过程了

# 总结 #

本篇文章主要是讲解了zk服务启动的入口，以及解析zoo.cfg，最后把其封装为QuorumPeerConfig，然后又创建了QuorumPeer, 然后会启动一个选举算法，选举算法是FastLeaderElection，同时别忘了还创建了一个ServerCnxnFactory，默认是NIOServerCnxnFactory，zk集群内部的通信组件

知识扩展：

在zk集群中主要涉及到集中连接

1、当leader选举时，是基于socket bio的方式进行通信选举leader的

2、zk集群之间的通信是基于nio长连接然后进行数据的同步通信的

3、客户端和zk服务也是基于nio进行长连接交互的

所以接下来可能会出两篇有关bio和nio的文章，会通过linux操作系统，来看看结合操作系统底层是如何进行通信的

[zookeeper_zookeeper]: https://blog.csdn.net/liereli/article/details/107588577
[20200727221832768.png]: /images/20230208/57643b8d031e4e4db94744f6a1c535c7.png
[20200727222257906.png]: /images/20230208/52f076b9906d42cc884d8f36efd309bf.png
[20200726062944835.png]: /images/20230208/74062103a6354cbeb73d425a066726ac.png
[20200727222853880.png]: /images/20230208/84398789aff94b3a9305fe07ea4e76cd.png