[TOC] **本文思维导图**  ### 前言 为了高可用和数据安全起见，zk集群一般都是由几个节点构成(由n/2+1，投票机制决定，肯定是奇数个节点)。多节点证明它们之间肯定会有数据的通信，同时，为了能够使zk集群对外是透明的，一个整体对外提供服务，那么客户端访问zk服务器的数据肯定是要数据同步，也即**数据一致性**。 zk集群是Leader/Follower模式来保证数据同步的。整个集群同一时刻只能有一个Leader，其他都是Follower或Observer。Leader是通过选举选出来的，这里涉及到ZAB协议(原子消息广播协议)。 ### 1.ZAB协议 #### 1.1 概念理解 为了更好理解下文，先说ZAB协议，它是选举过程和数据写入过程的基石。ZAB的核心是定义会改变zk服务器数据状态的事务请求的处理方式。 ZAB的理解：所有事务请求是由一个全局唯一的服务器来协调处理，这个的服务器就是Leader服务器， 其它服务器都是Follower服务器或Observer服务器。Leader服务器负责将一个客户端的请求转换成那个一个**事务Proposalͧ(提议)**，将该Proposal分发给集群中所有的Follower服务器。然后Leader服务器需要等待所有Follower服务器的应答，当Leader服务器收到超过**半数**的Follower服务器进行了明确的应答后，Leader会再次向所有的Follower服务器分发Commit消息，要求其将前一个Proposal进行提交。 注意**事务提议**这个词，就类似 **人大代表大会提议** ，提议就代表会有应答，之间有通信。因此在zk的ZAB协议为了可靠性和可用性，会有**投票**，**应答**等操作来保证整个zk集群的正常运行。 总的来说就是，涉及到客户端对zk集群数据改变的行为都先由Leader统一响应，然后再把请求转换为事务转发给其他所有的Follower，Follower应答并处理事务，最后再反馈。如果客户端只是读请求，那么zk集群所有的节点都可以响应这个请求。 #### 1.2 ZAB协议三个阶段 * 1.发现(选举Leader过程) * 2.同步(选出Leader后，Follower和Observer需进行数据同步) * 3.广播(同步之后，集群对外工作响应请求，并进行消息广播，实现数据在集群节点的副本存储) 下面会逐点分析，但是在这之前先来了解了解zookeeper服务器的知识吧。 ### 2.Zookeeper服务器 #### 2.1 zk服务器角色 * Leader * 事务请求的唯一调度和处理者，保证集群事务处理的顺序序性 * 集群内部各服务器的调度者 * Follower * 处理客户端非事务请求，转发事务请求给Leader服务器 * 参与事务请求Proposal的投票 * 参与Leader的选举投票 * Observer * 处理客户端非事务请求，转发事务请求给Leader服务器 * 不参加任何形式的投票，包括选举和事务投票(超过半数确认) * Observer的存在是为了提高zk集群对外提供读性能的能力 整个zk集群的角色作用如下图：  #### 2.2 zk服务器状态 * LOOKING * 寻找Leader状态 * 当服务器处于这种状态时，表示当前没有Leader，需要进入选举流程 * FOLLOWING * 从机状态，表明当前服务器角色是Follower * OBSERVING * 观察者状态，表明当前服务器角色是Observer * LEADING * 领导者状态，表明当前服务器角色是Leader * ServerState 类维护服务器四种状态。  zk服务器的状态是随着机器的变化而变化的。比如Leader宕机了，服务器状态就变为LOOKING，通过选举后，某机器成为Leader，服务器状态就转换为LEADING。其他情况类似。 #### 2.3 zk服务器通信 集群嘛，节点之间肯定是要通信的。zokeeper通信有两个特点： * 1.使用的通信协议是**TCP协议**。在集群中到底是怎么连接的呢？还记得在配置zookeeper时要创建一个data目录并在其他创建一个myid文件并写入唯一的数字吗？zk服务器的TCP连接方向就是依赖这个myid文件里面的数字大小排列。数小的向数大的发起TCP连接。比如有3个节点，myid文件内容分别为1,2,3。zk集群的tcp连接顺序是1向2发起TCP连接，2向3发起TCP连接。如果有n个节点，那么tcp连接顺序也以此类推。这样整个zk集群就会连接起来。 * 2.zk服务器是多端口的。例如配置如下： ~~~jsx tickTime=2000 dataDir=/home/liangjf/app/zookeeper/data dataLogDir=/home/liangjf/app/zookeeper/log clientPort=2181 initLimit=5 syncLimit=2 server.1=192.168.1.1:2888:3888 server.2=192.168.1.2:2888:3888 server.3=192.168.1.3:2888:3888 ~~~ * 第1个端口是通信和数据同步端口，默认是2888 * 第2个端口是投票端口，默认是3888 ### 3.选举机制 #### 3.1 选举算法 从zookeeper开始发布以来，选举的算法也慢慢优化。现在为了可靠性和高可用，从3.4.0版本开始zookeeper只支持基于**Tcp**的**FastLeaderElection**选举协议。 * LeaderElection * Udp协议 * AuthFastLeaderElection * udp * **FastLeaderElection** * Udp * **Tcp** **FastLeaderElection选举协议**使用TCP实现**Leader投票选举算法**。它使用了类对象`quorumcnxmanager`管理连接。该算法是基于推送的，可以通过调节参数来改变选举的过程。第一，`finalizewait`决定等到决定Leader的时间。这是Leader选举算法的一部分。 `final static int finalizeWait = 200;`（选举Leader过程的进程时间） `final static int maxNotificationInterval = 60000;`(通知检查选中Leader的时间间隔) `final static int IGNOREVALUE = -1;` 这里先不详细分析，下面**3.5 选举算法源码分析及举栗子**才分析。 #### 3.2 何时触发选举 选举Leader不是随时选举的，毕竟选举有产生大量的通信，造成网络IO的消耗。因此下面情况才会出现选举： * 集群启动 * 服务器处于寻找Leader状态 * 当服务器处于LOOKING状态时，表示当前没有Leader，需要进入选举流程 * 崩溃恢复 * Leader宕机 * 网络原因导致过半节点与Leader心跳中断 #### 3.3 如何成为Leader * 数据新旧程度 * 只有拥有最新数据的节点才能有机会成为Leader * 通过zxid的大小来表示数据的新，zxid越大代表数据越新 * myid * 集群启动时，会在data目录下配置myid文件，里面的数字代表当前zk服务器节点的编号 * 当zk服务器节点数据一样新时， myid中数字越大的就会被选举成ОLeader * 当集群中已经有Leader时，新加入的节点不会影响原来的集群 * 投票数量 * 只有得到集群中多半的投票，才能成为Leader * 多半即：n/2+1,其中n为集群中的节点数量 #### 3.4 重要的zxid 由3.3知道zxid是判断能否成为Leader的条件之一，它代表服务器的数据版本的新旧程度。 zxid由两部分构成：主进程周期epoch和事务单调递增的计数器。zxid是一个64位的数，高32位代表**主进程周期epoch**，低32位代表**事务单调递增的计数器**。 **主进程周期epoch**也叫epoch，是选举的轮次，每选举一次就递增1。**事务单调递增的计数器**在每次选举完成之后就会从0开始。 如果是比较数据新旧的话，直接比较就可以了。因为如果是主进程周期越大，即高32位越大，那么低32位就不用再看了。如果主进程周期一致，低32位越大，整个zxid就越大。所以直接比较整个64位就可以了，不必高32位于高32位对比，低32位与低32位比较。 #### 3.5 选举算法源码分析及举栗子 ##### 3.5.1 举栗子 zookeeper选举有两种情况： * 1.集群首次启动 * 2.集群在工作时Leader宕机 选主原则如下(在选举时，对比次序是从上往下) * 1.`New epoch is higher` * 主周期更大，代所有一切是最新，就成为leader * 2.`New epoch is the same as current epoch, but new zxid is higher` * 主周期一致就是在同一轮选票中，zxid越大就成为leader，因为数据更新 * 3.`New epoch is the same as current epoch, new zxid is the same as current zxid, but server id is higher` * 主周期和zxid一致，就看机器的id(myid)，myid越大就成为leader 同时，在选举的时候是投票方式进行的，除主进程周期外，投票格式为(myid,zxid)。 第一种情况，比较容易理解，下面以3台机器为例子。 * 三个zk节点A，B，C，三者开始都没有数据，即Zxid一致，对应的myid为1，2，3。 * A启动myid为1的节点，zxid为0，此时只有一台服务器无法选举出Leader * B启动myid为2的节点，zxid为0，B的zxid与A一样，比较myid，B的myid为2比A为1大，B成Leader * C启动myid为3的节点，因为已经有Leader节点，则C直接加入集群，承认B是leader 第二种情况，已5台机器为例子。 * 五个节点A，B，C，D，E，B是Leader，其他是Follower，myid分别为1，2，3，4，5，zxid分别为3，4，5，6，6。运行到某个时刻时A，B掉线或宕机，此时剩下C D E。在同一轮选举中，C，D，E分别投自己和交叉投票。 * 第一次投票，都是投自己。 * 投票情况为：C:(3,5) D:(4,6) E:(5,6)。 * 同时也会收到其他机器的投票。 * 投票情况为：C:(3,5)(4,6)(5,6)，D:(4,6)(3,5)(5,6)，E:(5,6)(4,6)(3,5) * 机器内部会根据选主原则对比投票，变更投票，投票情况为：C:(3,5)(4,6)(5,6)【不变更】。 D:(4,6)(4,6)(5,6)【变更】。E:(5,6)(5,6)(5,6)【变更】 * 统计票数，C-1票，D-3票，E-5票。因此E成为Leader。 接下来就是对新Leader节点的检查，数据同步，广播，对外提供服务。 ##### 3.5.1 选举算法源码分析 选举算法的全部代码在`FastLeaderElection`类中。其他的`lookForLeader`函数是选举Leader的入口函数。 ~~~java //每一轮选举就会增大一次逻辑时钟，同时更新事务 synchronized(this){ logicalclock++; updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch()); } ~~~ //一直循环选举直到找到leader,这里把打印和不相关的都删除了，方便分析。 ~~~php while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){ //从通知队列拉取一个投票通知 Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); if(n == null){ //看是否选举时通知发送/接收超时 int tmpTimeOut = notTimeout*2; notTimeout = (tmpTimeOut < maxNotificationInterval? tmpTimeOut : maxNotificationInterval); } else if(self.getVotingView().containsKey(n.sid)) { switch (n.state) { case LOOKING://只有zk服务器状态为LOOKING时才会进行选举 // If notification > current, replace and send messages out if (n.electionEpoch > logicalclock) { //如果选举时的逻辑时钟大于发送通知来源的机器的逻辑时钟，就把对方的修改为自己的。 logicalclock = n.electionEpoch; recvset.clear(); //并统计票数，如果能成为leader就更新事务 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) { //如果通知来演的机器的逻辑时钟比本次我的选举时钟低，直接返回，什么都不做。因为对方没机会成为leader if(LOG.isDebugEnabled()){ LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x" + Long.toHexString(n.electionEpoch) + ", logicalclock=0x" + Long.toHexString(logicalclock)); } break; } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)) { //如果Epoch一样，就看zxid的比较。不过还是会更新事务和回传通知 updateProposal(n.leader, n.zxid, n.peerEpoch); sendNotifications(); } //把所有接收到的投票信息都放到recvset集合 recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); //统计谁的投票超过半数，就成为leader if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) { //验证一下，被选举的leader是否有变化，就是看符不符合 while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null){ if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, proposedLeader, proposedZxid, proposedEpoch)){ //符合就放进recvqueue集合 recvqueue.put(n); break; } } //改变选举为leader的机器的状态为LEADING if (n == null) { self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(proposedLeader, proposedZxid, proposedEpoch); leaveInstance(endVote); return endVote; } } break; case FOLLOWING: case LEADING: //在同一轮选举中，判断所有的通知，并确认自己是leader if(n.electionEpoch == logicalclock){ recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch)); if(termPredicate(recvset, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state)) && checkLeader(outofelection, n.leader, n.electionEpoch)) { self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch); leaveInstance(endVote); return endVote; } } //在对外提供服务前，先广播一次自己是leader的消息给所有follower，让大家认同我为leader。 outofelection.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state)); if (termPredicate(outofelection, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state)) && checkLeader(outofelection, n.leader, n.electionEpoch)) { synchronized(this){ logicalclock = n.electionEpoch; self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState()); } Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch); leaveInstance(endVote); return endVote; } break; } } } ~~~ 比较重要的子函数有以下这些： * 1.totalOrderPredicate。(投票比较变更原则，选举的核心) * * * ~~~java 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; } //按照这样的顺序比较优先：Epoch > Zxid > myid return ((newEpoch > curEpoch) || ((newEpoch == curEpoch) && ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId))))); } ~~~ * 2.termPredicate。(最终的计算票数。先把投票放到集合中，然后再统计。集合能去重) * * * ~~~kotlin private boolean termPredicate( HashMap<Long, Vote> votes, Vote vote) { HashSet<Long> set = new HashSet<Long>(); for (Map.Entry<Long,Vote> entry : votes.entrySet()) { if (vote.equals(entry.getValue())){ set.add(entry.getKey()); } } return self.getQuorumVerifier().containsQuorum(set); } ~~~ * 3.Messenger。(构造Messenger的时候创建2条线程WorkerSender和WorkerReceiver用于整个选举的集群投票通信) * * * ~~~cpp 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(); } ~~~ 其他细节不多说了，主要是sendqueue和recvqueue队列存放待发送投票通知和接收投票通知，WorkerSender和WorkerReceiver两条线程用于投票的通信，QuorumCnxManager manager用于真正和其他机器的tcp连接维护管理，Messenger是整个投票通信的管理者。 ### 3.数据同步机制 #### 3.1 同步准备 完成选举之后，为了数据一致性，需要进行数据同步流程。 ##### 3.1.1 Leader准备 * Leader告诉其它follower当前最新数据是什么即zxid * Leader构建一个NEWLEADER的包，包括当前最大的zxid，发送给所有的follower或者Observer * Leader给每个follower创建一个线程LearnerHandler来负责处理每个follower的数据同步请求，同时主线程开始阻塞，等到超过一半的follwer同步完成，同步过程才完成，leader才真正成为leader ##### 3.1.2 Follower准备 * 选举完成后，尝试与leader建立同步连接，如果一段时间没有连接上就报连接超时，重新回到选举状态FOLLOWING * 向leader发送FOLLOWERINFO包，带上follower自己最大的zxid #### 3.2 同步初始化 同步初始化涉及到三个东西：minCommittedLog、maxCommittedLog、zxid – minCommittedLog:最小的事务日志id，即zxid没有被快照存储的日志文件的第一条，每次快照存储 完，会重新生成一个事务日志文件 – maxCommittedLog:事务日志中最大的事务，即zxid ### 4.数据同步场景 * 直接差异化同步(DIFF同步) * 仅回滚同步TRUNCͨ，即删除多余的事务日志，比如原来的Leader宕机后又重新加入，可能存在它自己写 入提交但是别的节点还没来得及提交 * 先回滚再差异化同步(TRUNC+DIFF同步) * 全量同步(SNAP同步) 不同的数据同步算法适用不同的场景。 ### 5.广播流程 * 集群选举完成，并且完成数据同步后，开始对外服务，接收读写请求 * 当leader接收到客户端新的事务请求后，会生成对新的事务proposal，并根据zxid的顺序向所有的 follower分发事务proposal * 当follower收到leader的proposal时，根据接收的先后顺序处理proposal * 当Leader收到follower针对某个proposal过半的ack后，则发起事务提交，重新发起一个commit的 proposal * Follower收到commit的proposal后，记录事务提交，并把数据更新到内存数据库 * 补充说明 * 由于只有过半的机器给出反馈，则可能存在某时刻某些节点数据不是最新的 * 如果需要确定读取到的数据是最新的，则可以在读取之前，调用sync方法进行数据同步 ### 6.小结 在zookeeper中，除了watcher机制，会话管理，最重要的就是选举了。它是zookeeper集群的核心，也是广泛应用在商业中的前提。 作者：dandan的微笑 链接：https://www.jianshu.com/p/57fecbe70540 来源：简书