上篇文章 手把手带你撸zookeeper源码-zookeeper通信序列化协议，简单说了一下zookeeper的jute序列化通信协议，写了一个简单的demo，以及如何避免粘包和拆包的，也顺带写了一点有关dubbo的序列化协议和如何自己自定义序列化协议

回到上上篇文章 手把手带你撸zookeeper源码-zookeeper确定好角色后会做什么？主要分析了zookeer中leader启动的时候都会做些什么事，分析到了leader启动的时候会对server.x=zk1:2888:3888，中的2888端口进行监听，等待其他的follower来进行链接，如果在等待了ticket * initLimit时间之内还有follower还没和leader进行链接，leader便不再等待，就直接对外提供服务

这篇文章主要分析一下follower启动的时候都会做什么事情？如何向leader发起注册链接的，如何从leader进行数据同步的？

回到QuorumPeer中的run方法，找到如下代码

                case FOLLOWING: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;

这段代码就是当选举完leader之后，zk知道直接时follower角色之后，会进行follower角色的一系列初始化操作，我们先进入makeFollower方法

    protected Follower makeFollower(FileTxnSnapLog logFactory) throws IOException {return new Follower(this, new FollowerZooKeeperServer(logFactory, this,new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));}protected Leader makeLeader(FileTxnSnapLog logFactory) throws IOException {return new Leader(this, new LeaderZooKeeperServer(logFactory,this,new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));}protected Observer makeObserver(FileTxnSnapLog logFactory) throws IOException {return new Observer(this, new ObserverZooKeeperServer(logFactory,this, new ZooKeeperServer.BasicDataTreeBuilder(), this.zkDb));}

我这里一下子粘贴了三个方法，第一个makeFollower方法，就是在当前启动的角色是follower进行实例化一个Follower对象，第二个方法makeLeader之前分析过，就是当前启动的角色是leader时进行实例化一个Leader对象，第三个makeObserver方法，就是当前启动的角色是Observer时进行实例化一个Observer对象，这块唯一不同的地方就是，传递的ZookeeperServer的子类不一样，我们先看一下ZookeeperServer的类UML图

Leader角色创建了LeaderZookeeperServer

follower角色创建了FollowerZookeeperSever

observer角色创建了ObserverZookeeperServer

这个是很关键的点，因为在不同的zk角色启动的时候会调用setupRequestProcessors()这个方法，然后初始化自己的调用链processor，这块我们之后再分析

初始化完Follower对象之后，接下来会调用Follower.followerLeader的方法

// 查找leader所在服务器QuorumServer leaderServer = findLeader();            try {//向leader发起连接connectToLeader(leaderServer.addr, leaderServer.hostname);//向leader进行注册, 经过三次握手long newEpochZxid = registerWithLeader(Leader.FOLLOWERINFO);long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);if (newEpoch < self.getAcceptedEpoch()) {LOG.error("Proposed leader epoch " + ZxidUtils.zxidToString(newEpochZxid)+ " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));throw new IOException("Error: Epoch of leader is lower");}syncWithLeader(newEpochZxid);QuorumPacket qp = new QuorumPacket();while (this.isRunning()) {readPacket(qp); //从leader读数据processPacket(qp);}} catch (Exception e) {}

这个方法会做一些列的事情，我们一步步分析

第一件事

QuorumServer leaderServer = findLeader();  

根据自己本地zoo.cfg中配置的server.x和保存leader sid做对比，然后获取到leader server地址

//向leader发起连接
connectToLeader(leaderServer.addr, leaderServer.hostname);

找到了leader server之后就开始向leader发起链接请求

protected void connectToLeader(InetSocketAddress addr, String hostname)throws IOException, ConnectException, InterruptedException {sock = new Socket();        sock.setSoTimeout(self.tickTime * self.initLimit);// 最多可以重复连接5次，如果超过5此都没连接成功, 则当前follower放弃和leader进行连接for (int tries = 0; tries < 5; tries++) {try {sock.connect(addr, self.tickTime * self.syncLimit);sock.setTcpNoDelay(nodelay);break;} catch (IOException e) {if (tries == 4) {LOG.error("Unexpected exception",e);throw e;} else {LOG.warn("Unexpected exception, tries="+tries+", connecting to " + addr,e);sock = new Socket();sock.setSoTimeout(self.tickTime * self.initLimit);}}Thread.sleep(1000);}self.authLearner.authenticate(sock, hostname);// 通过socket获取输入输出流，并包装为jute序列化协议对象,// 接下来可以直接通过jute进行数据的序列化、反序列化读取数据和发送数据leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream()));bufferedOutput = new BufferedOutputStream(sock.getOutputStream());leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);}  

这个方法就是向leader发起链接的代码，就是socket之间的链接，每个follower会发起5此向leader链接的请求，如果一直失败，超过5此以后，就不再向发起链接了

leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream()));
bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
leaderOs = BinaryOutputArchive.getArchive(bufferedOutput);

最后的代码就是如果和leader链接成功之后，会把socket中的输入输出流封装为jute序列化协议对象leaderIs读数据和leaderOs写数据，之后如果要发送数据或者读取数据直接拿来使用即可

接下来registerWithLeader()方法会向leader发起三次握手建立一个following链接

protected long registerWithLeader(int pktType) throws IOException{long lastLoggedZxid = self.getLastLoggedZxid();QuorumPacket qp = new QuorumPacket();                qp.setType(pktType);qp.setZxid(ZxidUtils.makeZxid(self.getAcceptedEpoch(), 0));/** Add sid to payload* 包装当前的sid为learnerInfo对象*/LearnerInfo li = new LearnerInfo(self.getId(), 0x10000);ByteArrayOutputStream bsid = new ByteArrayOutputStream();BinaryOutputArchive boa = BinaryOutputArchive.getArchive(bsid);boa.writeRecord(li, "LearnerInfo");qp.setData(bsid.toByteArray());//发送sid和协议版本号给leader,writePacket(qp, true);readPacket(qp);        final long newEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());// 第一次接收到leader发送过来的数据，leader发送过来的协议版本号if (qp.getType() == Leader.LEADERINFO) {// we are connected to a 1.0 server so accept the new epoch and read the next packetleaderProtocolVersion = ByteBuffer.wrap(qp.getData()).getInt();byte epochBytes[] = new byte[4];final ByteBuffer wrappedEpochBytes = ByteBuffer.wrap(epochBytes);if (newEpoch > self.getAcceptedEpoch()) {wrappedEpochBytes.putInt((int)self.getCurrentEpoch());self.setAcceptedEpoch(newEpoch);} else if (newEpoch == self.getAcceptedEpoch()) {wrappedEpochBytes.putInt(-1);} else {throw new IOException("Leaders epoch, " + newEpoch + " is less than accepted epoch, " + self.getAcceptedEpoch());}QuorumPacket ackNewEpoch = new QuorumPacket(Leader.ACKEPOCH, lastLoggedZxid, epochBytes, null);// 发送ackwritePacket(ackNewEpoch, true);return ZxidUtils.makeZxid(newEpoch, 0);} else {if (newEpoch > self.getAcceptedEpoch()) {self.setAcceptedEpoch(newEpoch);}if (qp.getType() != Leader.NEWLEADER) {LOG.error("First packet should have been NEWLEADER");throw new IOException("First packet should have been NEWLEADER");}return qp.getZxid();}} 

有两种情况，如果zookeeper刚刚启动或者是刚新加入进来的follower，本地没有数据。第二种就是集群中某一台宕机了，本地中保存的有自己最新的zxid和可以接受的acceptEpoch(epoch之前有提到过，就是一旦有leader变更，epoch就会加一)

上次握手大概时这样子的

第一次: follower启动的时候，会根据epoch来生成一个zxid和epoch(leader版本号)，还有sid以及协议版本号(0x10000)发送给leader

第二次: follower接收到leader返回回来的信息，然后读取leader发送回来的epoch(当前leader的版本)，如果leader发送过来的epoch比自己本地的epoch要大，则更新自己本地的epoch，和leader保持同步，如果时相等的，那么最后一步的ack可以不用再次发送epoch给leader，如果leader发送过来的epoch小于本地的epoch，则会抛出异常，此时有可能因为zookeeper集群脑裂，导致链接上了一个版本较低的leader

第三次: follower会发送一个ack给leader，把自己最新的zxid发送给leader

才想一下最终把zxid发送给leader干啥？

我们可以想想，如果我当前的zk是从故障中恢复的，那么此时我本地的数据肯定时落后于leader的，我把我的zxid发送给leader，那么leader会根据自己的数据来判断一下有多少数据需要同步给我，需不需要全量同步？

其实zk故障重启之后如何同步数据有好几种情况，我们在下面代码中来详细说

回到Follower.followerLeader()方法中，我们继续下面的代码

// 校验一下leader的zxid是否小于我们的, 这种情肯定不会发生，只是做个安全检查long newEpoch = ZxidUtils.getEpochFromZxid(newEpochZxid);if (newEpoch < self.getAcceptedEpoch()) {LOG.error("Proposed leader epoch " + ZxidUtils.zxidToString(newEpochZxid)+ " is less than our accepted epoch " + ZxidUtils.zxidToString(self.getAcceptedEpoch()));throw new IOException("Error: Epoch of leader is lower");}

这块代码上面写了注释, 就是做一个安全检查

又一个关键代码来了

syncWithLeader(newEpochZxid);

看方法名字: 同步leader，我们看看代码，上上面的代码我们把本地的最新zxid发送到leader了

protected void syncWithLeader(long newLeaderZxid) throws IOException, InterruptedException{QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);QuorumPacket qp = new QuorumPacket();long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);boolean snapshotNeeded = true;readPacket(qp);LinkedList<Long> packetsCommitted = new LinkedList<Long>();LinkedList<PacketInFlight> packetsNotCommitted = new LinkedList<PacketInFlight>();synchronized (zk) {// 如果和leader之间有不同的数据, 则不需要快照if (qp.getType() == Leader.DIFF) {LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));// 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可snapshotNeeded = false;}// 从leader获取快照信息else if (qp.getType() == Leader.SNAP) {LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));// The leader is going to dump the database// clear our own database and read// 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化// 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据zk.getZKDatabase().clear();zk.getZKDatabase().deserializeSnapshot(leaderIs);// 读取签名String signature = leaderIs.readString("signature");if (!signature.equals("BenWasHere")) {LOG.error("Missing signature. Got " + signature);throw new IOException("Missing signature");                   }zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());} else if (qp.getType() == Leader.TRUNC) {//we need to truncate the log to the lastzxid of the leader// 根据leader的lastzxid对本地日志进行截断// 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower// 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据// 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的// 所以会把宕机之前最后的一条数据给删除掉LOG.warn("Truncating log to get in sync with the leader 0x"+ Long.toHexString(qp.getZxid()));boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());if (!truncated) {System.exit(13);}zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());}else {LOG.error("Got unexpected packet from leader "+ qp.getType() + " exiting ... " );System.exit(13);}// 创建会话跟踪器zk.createSessionTracker();long lastQueued = 0;boolean writeToTxnLog = !snapshotNeeded;// we are now going to start getting transactions to apply followed by an UPTODATEouterLoop:while (self.isRunning()) {readPacket(qp);switch(qp.getType()) {case Leader.PROPOSAL:PacketInFlight pif = new PacketInFlight();pif.hdr = new TxnHeader();pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);if (pif.hdr.getZxid() != lastQueued + 1) {LOG.warn("Got zxid 0x"+ Long.toHexString(pif.hdr.getZxid())+ " expected 0x"+ Long.toHexString(lastQueued + 1));}lastQueued = pif.hdr.getZxid();packetsNotCommitted.add(pif);break;case Leader.COMMIT:if (!writeToTxnLog) {pif = packetsNotCommitted.peekFirst();if (pif.hdr.getZxid() != qp.getZxid()) {LOG.warn("Committing " + qp.getZxid() + ", but next proposal is " + pif.hdr.getZxid());} else {zk.processTxn(pif.hdr, pif.rec);packetsNotCommitted.remove();}} else {packetsCommitted.add(qp.getZxid());}break;case Leader.INFORM:PacketInFlight packet = new PacketInFlight();packet.hdr = new TxnHeader();packet.rec = SerializeUtils.deserializeTxn(qp.getData(), packet.hdr);lastQueued = packet.hdr.getZxid();if (!writeToTxnLog) {// Apply to db directly if we haven't taken the snapshotzk.processTxn(packet.hdr, packet.rec);} else {packetsNotCommitted.add(packet);packetsCommitted.add(qp.getZxid());}break;case Leader.UPTODATE:   // 标识同步leader数据已经同步完了，可以接收客户端发起的请求了if (isPreZAB1_0) {zk.takeSnapshot();self.setCurrentEpoch(newEpoch);}self.cnxnFactory.setZooKeeperServer(zk);                break outerLoop;    // 跳出循环case Leader.NEWLEADER: File updating = new File(self.getTxnFactory().getSnapDir(),QuorumPeer.UPDATING_EPOCH_FILENAME);if (!updating.exists() && !updating.createNewFile()) {throw new IOException("Failed to create " +updating.toString());}if (snapshotNeeded) {zk.takeSnapshot();}self.setCurrentEpoch(newEpoch);if (!updating.delete()) {throw new IOException("Failed to delete " +updating.toString());}writeToTxnLog = true; //Anything after this needs to go to the transaction log, not applied directly in memoryisPreZAB1_0 = false;writePacket(new QuorumPacket(Leader.ACK, newLeaderZxid, null, null), true);break;}}}
}

这块代码有点多，我也只是截取了一部分，今天主要分析一下上面的代码，这个方法截取到了487行，剩余的代码我们之后分析

QuorumPacket ack = new QuorumPacket(Leader.ACK, 0, null, null);QuorumPacket qp = new QuorumPacket();long newEpoch = ZxidUtils.getEpochFromZxid(newLeaderZxid);boolean snapshotNeeded = true;readPacket(qp);

这块很简单，就是从leader中读取发送过来的数据

if (qp.getType() == Leader.DIFF) {LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));// 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可snapshotNeeded = false;}// 从leader获取快照信息else if (qp.getType() == Leader.SNAP) {LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));// The leader is going to dump the database// clear our own database and read// 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化// 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据zk.getZKDatabase().clear();zk.getZKDatabase().deserializeSnapshot(leaderIs);// 读取签名String signature = leaderIs.readString("signature");if (!signature.equals("BenWasHere")) {LOG.error("Missing signature. Got " + signature);throw new IOException("Missing signature");                   }zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());} else if (qp.getType() == Leader.TRUNC) {//we need to truncate the log to the lastzxid of the leader// 根据leader的lastzxid对本地日志进行截断// 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower// 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据// 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的// 所以会把宕机之前最后的一条数据给删除掉LOG.warn("Truncating log to get in sync with the leader 0x"+ Long.toHexString(qp.getZxid()));boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());if (!truncated) {// not able to truncate the logLOG.error("Not able to truncate the log "+ Long.toHexString(qp.getZxid()));System.exit(13);}zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());}else {LOG.error("Got unexpected packet from leader "+ qp.getType() + " exiting ... " );System.exit(13);}

大家也可以看着我写的注释进行理解，几个分支我们单独分析一下

// 如果和leader之间有不同的数据, 则不需要快照if (qp.getType() == Leader.DIFF) {LOG.info("Getting a diff from the leader 0x{}", Long.toHexString(qp.getZxid()));// 如果当前follower宕机，然后恢复重启，此时会落后leader一部分数据，然后去同步宕机之后的数据即可snapshotNeeded = false;}

这个分支的情况就是如果zk集群中某个follower宕机了，然后进行了重启，那么此时他直接取leader中取同步它宕机期间未同步的数据即可，同步数据在下面的代码中

if (qp.getType() == Leader.SNAP) {LOG.info("Getting a snapshot from leader 0x" + Long.toHexString(qp.getZxid()));// The leader is going to dump the database// clear our own database and read// 清空本地的数据文件和内存数据库，从leader中读取快照数据，进行反序列化// 可能当前的zk服务器新加入集群的，此时会从leader进行全量同步数据zk.getZKDatabase().clear();zk.getZKDatabase().deserializeSnapshot(leaderIs);// 读取签名String signature = leaderIs.readString("signature");if (!signature.equals("BenWasHere")) {LOG.error("Missing signature. Got " + signature);throw new IOException("Missing signature");                   }zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());}

第二个分支，这应该有两种情况下会发生，第一个情况就是如果在现有zk集群中新加入进来follower或者observer，那么此时它们刚开始是没有数据的，此时肯定会进行全量的同步，直接把leader中的快照全部同步过来。另一个情况有可能是这样的，就是leader判断当前follower和leader之间的数据相差太大，评估一下增量同步比快照全量同步的代价要大，也可能会触发全量快照同步，这块代码主要是先清空一下本地的内存数据库

if (qp.getType() == Leader.TRUNC) {//we need to truncate the log to the lastzxid of the leader// 根据leader的lastzxid对本地日志进行截断// 有这样的一种场景，当前zk服务器原来是leader，然后有客户端发送过来数据，写入本地日志文件，还没来得及发送给follower// 此时leader宕机了，然后原有的集群中某个follower会被选为leader，此时相当于就丢掉了一条数据// 当挂掉的leader宕机恢复之后，会作为一个follower加入集群中，此时回和leader进行同步，因为此时的leader中是没有这条数据的// 所以会把宕机之前最后的一条数据给删除掉LOG.warn("Truncating log to get in sync with the leader 0x"+ Long.toHexString(qp.getZxid()));boolean truncated=zk.getZKDatabase().truncateLog(qp.getZxid());if (!truncated) {// not able to truncate the logLOG.error("Not able to truncate the log "+ Long.toHexString(qp.getZxid()));System.exit(13);}zk.getZKDatabase().setlastProcessedZxid(qp.getZxid());}

第三种情况很特殊，只有是leader挂了之后才会出现这种情况

大家知道zk集群写数据保持了顺序一致性，并且是2PC的过半写机制，此时就导致了这种情况

比方说，现在客户端写入一条数据，然后请求最终会转发给leader，leader回先把数据写入到本地日志文件，然后在发送proposal给其他的follower，但是在写入本地磁盘成功之后，还没来得及发送给其他的follower proposal消息，忽然leader宕机或者网络出现了问题导致其他的follower认为leader宕机了。因为zk集群肯定是会从follower中重新选举一个leader，而所有的follower都没有收到这个proposal。此时新的leader选举出来了。旧的leader宕机之后可能运维重启，也可能网络恢复了。此时它发现集群中已经有了leader，那么它会作为一个follower节点加入到集群中取。此时follower会把自己的最新zxid发送给leader，leader发现自己根本就没有这个zxid，然后旧leader必须把这个zxid对应的数据从日志和内存中给删除掉，然后再继续从leader中同步因为宕机或者网络原因导致没有同步到的最新数据。大概就是这种情况，大家可以好好思考一下

这篇文章先写到这里吧，下篇文章再继续分析接下来的代码