Zookeeper原理分析之存储结构ZkDatabase

ZKDatabase在内存中维护了zookeeper的sessions， datatree和commit logs集合。 当zookeeper server启动的时候会将txnlogs和snapshots从磁盘读取到内存中。核心逻辑主要在方法ZkDatabase.loadDataBase()中实现，其代码如下：

    /**
     * 将数据从磁盘加载到内存中，并将事物添加到内存中的提交日志中
     * @return 磁盘上最后一个有效的zxid
     * @throws IOException
     */
    public long loadDataBase() throws IOException {
        PlayBackListener listener=new PlayBackListener(){
            public void onTxnLoaded(TxnHeader hdr,Record txn){
                Request r = new Request(null, 0, hdr.getCxid(),hdr.getType(),
                        null, null);
                r.txn = txn;
                r.hdr = hdr;
                r.zxid = hdr.getZxid();
                addCommittedProposal(r);
            }
        };
        
        long zxid = snapLog.restore(dataTree,sessionsWithTimeouts,listener);
        initialized = true;
        return zxid;
    }

loadDataBase的流程如下：

• 构建一个PlayBackListener对象
• snapshot的反序列，倒叙排目录下的snapshot文件，遍历查找出最新的那个有效snapshot文件进行反序列化到内存。snapshot的反序列后我们会知道snapshot最新的zxid叫做lastProcessedZxid, 这个lastProcessedZxid之前的事务操作，都成功执行并序列到snapshot中可恢复到内存，lastProcessedZxid之后的操作只有事务日志，不能直接通过snapshot恢复。

snapLog.restore实现如下：

    public long restore(DataTree dt, Map<Long, Integer> sessions, 
            PlayBackListener listener) throws IOException {
		//从最近的快照中对数据树进行反序列化
        snapLog.deserialize(dt, sessions);
		//lastProcessedZxid+1从事务日志文件txnLog读取事务操作
        FileTxnLog txnLog = new FileTxnLog(dataDir);
        TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1);
        long highestZxid = dt.lastProcessedZxid;
        TxnHeader hdr;
        try {
            while (true) {
                //遍历TxnIterator，执行processTransaction方法，就是把事务操作在内存中在执行一遍把丢失的操作补回来
                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);
                }
				// 同时将事务操作通过PlayBackListener添加到commitedLog集合，commitedLog的事务操作在服务恢复的时候会同步到其他leaner server, 因为很有可能其他leaner server也没有及时的takesnapshot
                listener.onTxnLoaded(hdr, itr.getTxn());
                if (!itr.next()) 
                    break;
            }
        } finally {
            if (itr != null) {
                itr.close();
            }
        }
	//返回最后的事务日志zxid给database，作为ZKDatabase的最新事物id
        return highestZxid;
    }

在zookeeperServer成功loadDatabase后，会及时主动的做一次takesnapshot操作来得到一份最新的内存影像。snapshot是内存数据的某个点一份影像，takeSnapshot操作还是很耗时，为了性能根据一下算法操作：

• 创建在同步处理器SyncRequestProcessor
• 100000/2 + random.nextInt(100000/2),这个十万是一个默认值可配置）计算出一个值，如果logCount大于这个值，就进行takeSnapshot操作

上面的算法存在一个问题， 那就是在非正常关机情况下，最新有效的那个snapshot并不是内存中最新的数据，所以需要利用txnLogs来把没有生成snapshot的操作在内存重新执行一边来恢复到非正常关闭服务那一刻内存情况。