手把手带你撸zookeeper源码-zookeeper故障重启时如何恢复数据(一)

雨点打透心脏的1/2处 2022-11-27 06:53 119阅读 0赞

在之前的文章 [手把手带你撸zookeeper源码-zookeeper中follower启动的时候会做什么?][zookeeper_-zookeeper_follower] 有分析过一部分follower启动时会调用syncWithLeader(zxid)方法, 此时方法会从leader中同步数据，但是回过头来看，感觉分析的不够深入，所以准备单独拉取出来一篇文章，来分析一下当follower启动时如何恢复数据的

其实当一个zookeeper进程启动加入现有集群时，会有以下几种情况:

1、当前zookeeper是新增的服务器, 然后作为observer角色加入集群

2、当前zookeeper是新增的服务器, 然后作为follower角色加入集群

3、当前zookeeper是故障节点恢复重启，原来是observer角色，现在重启加入集群

4、当前zookeeper是故障节点恢复重启，原来是follower角色，现在重启加入集群

5、当前zookeeper是故障节点恢复重启，原来是leader角色，现在重启以follower角色加入集群

以上五种情况都是需要考虑如何进行数据恢复的，其实他们都是大同小异，唯一的区别就是observer只从leader同步数据，不参与选举和2PC阶段提交, 而leader宕机重启，从leader变为follower加入集群时可能会导致数据的丢失，需要特殊处理。但是他们的相同点就是：第一、从本地磁盘日志文件中加载数据到内存中。第二、加载完毕之后和leader建立连接，和leader开始通信，并最终确定要同步哪些数据，增量同步、全量同步、截断 + 增量同步等不同的方式不同的处理，下面以follower为例来分析一下数据是如何恢复的

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

如上代码: 第一个点，就是我们启动zk时，其实就是执行QuorumPeerMain中的main方法，然后调用QuorumPeer.start()来启动一个zk节点，就是以上的代码，我们本篇文章主要分析数据的同步，所以其他无关的代码不讨论，此时关注loadDataBase()，这个方法就是从本地文件中读取日志文件来恢复数据的

private void loadDataBase() {
            File updating = new File(getTxnFactory().getSnapDir(),
                                     UPDATING_EPOCH_FILENAME);
    		
                zkDb.loadDataBase();
    
                // load the epochs
                long lastProcessedZxid = zkDb.getDataTree().lastProcessedZxid;
        		long epochOfZxid = ZxidUtils.getEpochFromZxid(lastProcessedZxid);
                try {
                	currentEpoch = readLongFromFile(CURRENT_EPOCH_FILENAME);
                    if (epochOfZxid > currentEpoch && updating.exists()) {
                        setCurrentEpoch(epochOfZxid);
                        if (!updating.delete()) {
                            throw new IOException("Failed to delete " +
                                                  updating.toString());
                        }
                    }
                } catch(FileNotFoundException e) {
                	currentEpoch = epochOfZxid;
                	writeLongToFile(CURRENT_EPOCH_FILENAME, currentEpoch);
                }
                if (epochOfZxid > currentEpoch) {
                	throw new IOException("The current epoch, " + ZxidUtils.zxidToString(currentEpoch) + ", is older than the last zxid, " + lastProcessedZxid);
                }
                	acceptedEpoch = readLongFromFile(ACCEPTED_EPOCH_FILENAME);
                if (acceptedEpoch < currentEpoch) {
                	throw new IOException("The accepted epoch, " + ZxidUtils.zxidToString(acceptedEpoch) + " is less than the current epoch, " + ZxidUtils.zxidToString(currentEpoch));
                }
    	}

我们详细分析一下如何加载的

我们首先要需要getTxnFactory().getSnapDir() 这个获取的目录是哪个目录下的文件呢？这需要往前看代码，就是初始书QuorumPeer对象的时候，会有下面的代码

quorumPeer.setTxnFactory(new FileTxnSnapLog(
                      new File(config.getDataLogDir()),
                      new File(config.getDataDir())));

来者是文件快照日志目录，这里的config.getDataLogDir()就是我们之前配置文件zoo.cfg中配置的dataLogDir目录，而config.getDataDir()就是zoo.cfg中配置的dataDir相应的目录，在FileTxnSnapLog构造函数中进行了赋值, snapDir = dataDir

File updating = new File(getTxnFactory().getSnapDir(),
                                     UPDATING_EPOCH_FILENAME);

第一行代码就是创建一个File对象，指向了dataDir目录

第二行代码zkDb.loadDataBase()方法，其中先找一下zkDb是在哪初始化的呢？

quorumPeer.setZKDatabase(new ZKDatabase(quorumPeer.getTxnFactory()));

还是quorumPeer对象初始化时进行创建的ZKDatabase, 即zk数据库，这个对象里面就保存了整个zk内存目录树以及节点数据

public ZKDatabase(FileTxnSnapLog snapLog) {
            dataTree = new DataTree();
            sessionsWithTimeouts = new ConcurrentHashMap<Long, Integer>();
            this.snapLog = snapLog;
        }

其中创建了一个DataTree对象，即目录树结构，有关zookeeper以什么样的方式保存数据的，以及使用什么数据结构我们可以单独再去分析

我们看一下zkDb.loadDataBase()方法中的代码

public long loadDataBase() throws IOException {
            long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);
            initialized = true;
            return zxid;
        }

主要的就是snapLog.restore()方法

public long restore(DataTree dt, Map<Long, Integer> sessions, 
                PlayBackListener listener) throws IOException {
            // 把snapLog文件中的数据反序列化到dt数据结构中
            snapLog.deserialize(dt, sessions);
            return fastForwardFromEdits(dt, sessions, listener);
        }

第一行代码就是从最近的一个快照文件中快速反序列化，把数据存入到DataTree数据结构中，只会找到最新的100个快照文件进行恢复，以前的快照文件进行放弃

public long deserialize(DataTree dt, Map<Long, Integer> sessions)
                throws IOException {
            // we run through 100 snapshots (not all of them)
            // if we cannot get it running within 100 snapshots
            // we should  give up
            List<File> snapList = findNValidSnapshots(100);
            if (snapList.size() == 0) {
                return -1L;
            }
            File snap = null;
            boolean foundValid = false;
            for (int i = 0; i < snapList.size(); i++) {
                snap = snapList.get(i);
                InputStream snapIS = null;
                CheckedInputStream crcIn = null;
                try {
                    LOG.info("Reading snapshot " + snap);
                    snapIS = new BufferedInputStream(new FileInputStream(snap));
                    crcIn = new CheckedInputStream(snapIS, new Adler32());
                    InputArchive ia = BinaryInputArchive.getArchive(crcIn);
                    deserialize(dt,sessions, ia);
                    long checkSum = crcIn.getChecksum().getValue();
                    long val = ia.readLong("val");
                    if (val != checkSum) {
                        throw new IOException("CRC corruption in snapshot :  " + snap);
                    }
                    foundValid = true;
                    break;
                } catch(IOException e) {
                    LOG.warn("problem reading snap file " + snap, e);
                } finally {
                    if (snapIS != null) 
                        snapIS.close();
                    if (crcIn != null) 
                        crcIn.close();
                } 
            }
            if (!foundValid) {
                throw new IOException("Not able to find valid snapshots in " + snapDir);
            }
            dt.lastProcessedZxid = Util.getZxidFromName(snap.getName(), SNAPSHOT_FILE_PREFIX);
            return dt.lastProcessedZxid;
        }

fastForwardFromEdits()方法

public long fastForwardFromEdits(DataTree dt, Map<Long, Integer> sessions,
                                         PlayBackListener listener) throws IOException {
            FileTxnLog txnLog = new FileTxnLog(dataDir);
            TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
            long highestZxid = dt.lastProcessedZxid;
            TxnHeader hdr;
            try {
                while (true) {
                    // iterator points to 
                    // the first valid txn when initialized
                    hdr = itr.getHeader();
                    if (hdr == null) {
                        //empty logs 
                        return dt.lastProcessedZxid;
                    }
                    if (hdr.getZxid() < highestZxid && highestZxid != 0) {
                        LOG.error("{}(higestZxid) > {}(next log) for type {}",
                                new Object[] { highestZxid, hdr.getZxid(),
                                        hdr.getType() });
                    } else {
                        highestZxid = hdr.getZxid();
                    }
                    try {
                        //处理提交的数据
                        processTransaction(hdr,dt,sessions, itr.getTxn());
                    } catch(KeeperException.NoNodeException e) {
                       throw new IOException("Failed to process transaction type: " +
                             hdr.getType() + " error: " + e.getMessage(), e);
                    }
                    listener.onTxnLoaded(hdr, itr.getTxn());
                    if (!itr.next()) 
                        break;
                }
            } finally {
                if (itr != null) {
                    itr.close();
                }
            }
            return highestZxid;
        }

这个方法主要是用来恢复在日志文件中，但是不在快照中的数据，把其恢复到DataTree数据结构中

以上只是在启动时先从本地文件中快速恢复数据，第一就是先把快照文件快速恢复，在后台有个线程会定期把内存中的数据序列化为一个快照文件，所以恢复数据是很快的。第二日志文件，每写一条数据就会写入到日志文件中，所以在把快照文件中的数据恢复到内存中之后，再去日志文件中找到快照文件生成之后但是还没来得及生成下次快照文件之前的数据，再恢复到内存中去，此时才算从文件中恢复数据完毕

当然，如果是新加入集群的zookeeper进程，此时本地肯定是没有数据文件的，当然也就没什么可恢复的了

接下来就是第二段，当从本地的快照文件 + 日志文件恢复完数据之后，有可能出现以下几种情况

1、如果是新加入集群的zookeeper，那需要从leader全量同步leader的数据，然后开始对外提供服务

2、如果是宕机恢复的zookeeper，则从本地恢复文件之后，需要去leader中同步宕机之后产生的最新数据

以上两种情况之前有简单分析过，但是不够详细，下篇文章会逐行代码一一分析

[zookeeper_-zookeeper_follower]: https://blog.csdn.net/liereli/article/details/107875454