从ZooKeeper源代码看如何实现分布式系统（二）数据的高可用存储

    这篇先从数据的高可用存储说起。ZooKeeper提供了分布式的目录服务，它存储的数据相比一个分布式存储系统来说很少，它主要是用来做分布式协同操作的。但是麻雀虽小，五脏俱全，ZooKeeper也必须要提供数据的高可用存储，对数据进行备份和恢复，以防出现服务器宕机导致数据丢失的情况。

    高可用的数据存储有一个比较通用的解决方案，就是数据文件 + 日志文件的方式。比如传统数据库中的数据文件 + undo/redo日志就可以来进行数据备份和恢复，在日志文件中加入检查点checkpoint，可以更加快速地进行数据的恢复。所以对于高可用的数据存储来说，我们要考察3个方面：

数据文件
日志文件
检查点
数据文件

ZooKeeper的数据文件采用快照文件的方式来记录和持久化运行时的数据。顶层接口是SnapShot，提供了对运行时的数据DataTree和session的序列化和反序列化操作。DataTree保存了运行时的数据。

public interface SnapShot {

long deserialize(DataTree dt, Map<Long, Integer> sessions)
throws IOException;

void serialize(DataTree dt, Map<Long, Integer> sessions,
File name)
throws IOException;

File findMostRecentSnapshot() throws IOException;

void close() throws IOException;
}

SnapShot的默认实现类是FileSnapShot，提供了把DataTree和Session持久化到文件的能力。来看一下它的序列化实现

1. 创建一个具备校验和的文件输出流

2. 对象的序列化采用apache jute框架，创建一个jute的OutputArchive的实现。下面给出了OutputArchive接口的定义，可以看到它和Thrift的TProtocol的定义基本一致，提供了一系列的write类型和read类型接口，是jute 序列化的顶层接口

3. OutputArchive的默认实现是BinaryOutputArchive，和Thrift的TBinaryProtocol实现基本一致，提供了二进制的序列化协议，内部采用DataOutputStream，把不同的数据类型写到Byte数组中

4. 快照文件的文件头对象FileHeader，包含一个魔数ZKSN, 版本号和dbId。 FileHeader实现了jute的Record接口，提供了serialize和deserialize方法实现

5. 快照文件体使用SerializeUtils这个辅助类来实现，先序列化Session，序列化Session时，先写一个Long类型的SessionId，再写一个int类型的timeout。再序列化DataTree，它也实现了Jute的Record类，实现了序列化自己的serialize方法

6. DataTree的serialize方法，先序列化ACL信息，再序列化DataTree中的DataNode，采用中序遍历的方式递归遍历DataTree的所有节点。最后写入"/"表示文件结尾

// FileSnapshot
public synchronized void serialize(DataTree dt, Map<Long, Integer> sessions, File snapShot)
throws IOException {
if (!close) {
OutputStream sessOS = new BufferedOutputStream(new FileOutputStream(snapShot));
CheckedOutputStream crcOut = new CheckedOutputStream(sessOS, new Adler32());
//CheckedOutputStream cout = new CheckedOutputStream()
OutputArchive oa = BinaryOutputArchive.getArchive(crcOut);
FileHeader header = new FileHeader(SNAP_MAGIC, VERSION, dbId);
serialize(dt,sessions,oa, header);
long val = crcOut.getChecksum().getValue();
oa.writeLong(val, "val");
oa.writeString("/", "path");
sessOS.flush();
crcOut.close();
sessOS.close();
}
}

 protected void serialize(DataTree dt,Map<Long, Integer> sessions,
            OutputArchive oa, FileHeader header) throws IOException {
        // this is really a programmatic error and not something that can
        // happen at runtime
        if(header==null)
            throw new IllegalStateException(
                    "Snapshot's not open for writing: uninitialized header");
        header.serialize(oa, "fileheader");
        SerializeUtils.serializeSnapshot(dt,oa,sessions);
    }

public class FileHeader implements Record {
  private int magic;
  private int version;
  private long dbid;
  public FileHeader() {
  }
 public void serialize(OutputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(this,tag);
    a_.writeInt(magic,"magic");
    a_.writeInt(version,"version");
    a_.writeLong(dbid,"dbid");
    a_.endRecord(this,tag);
  }
  public void deserialize(InputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(tag);
    magic=a_.readInt("magic");
    version=a_.readInt("version");
    dbid=a_.readLong("dbid");
    a_.endRecord(tag);
}

public interface Record {
    public void serialize(OutputArchive archive, String tag)
        throws IOException;
    public void deserialize(InputArchive archive, String tag)
        throws IOException;
}
public interface OutputArchive {
    public void writeByte(byte b, String tag) throws IOException;
    public void writeBool(boolean b, String tag) throws IOException;
    public void writeInt(int i, String tag) throws IOException;
    public void writeLong(long l, String tag) throws IOException;
    public void writeFloat(float f, String tag) throws IOException;
    public void writeDouble(double d, String tag) throws IOException;
    public void writeString(String s, String tag) throws IOException;
    public void writeBuffer(byte buf[], String tag)
        throws IOException;
    public void writeRecord(Record r, String tag) throws IOException;
    public void startRecord(Record r, String tag) throws IOException;
    public void endRecord(Record r, String tag) throws IOException;
    public void startVector(List v, String tag) throws IOException;
    public void endVector(List v, String tag) throws IOException;
    public void startMap(TreeMap v, String tag) throws IOException;
    public void endMap(TreeMap v, String tag) throws IOException;

}

public class BinaryOutputArchive implements OutputArchive {
    private ByteBuffer bb = ByteBuffer.allocate(1024);

    private DataOutput out;
    
    public static BinaryOutputArchive getArchive(OutputStream strm) {
        return new BinaryOutputArchive(new DataOutputStream(strm));
    }
    
    /** Creates a new instance of BinaryOutputArchive */
    public BinaryOutputArchive(DataOutput out) {
        this.out = out;
    }
    
    public void writeByte(byte b, String tag) throws IOException {
        out.writeByte(b);
    }
    
    public void writeBool(boolean b, String tag) throws IOException {
        out.writeBoolean(b);
    }
    
    public void writeInt(int i, String tag) throws IOException {
        out.writeInt(i);
    }
    
    public void writeLong(long l, String tag) throws IOException {
        out.writeLong(l);
    }
    
    public void writeFloat(float f, String tag) throws IOException {
        out.writeFloat(f);
    }
    
    public void writeDouble(double d, String tag) throws IOException {
        out.writeDouble(d);
    }

// SerializeUtils
 public static void serializeSnapshot(DataTree dt,OutputArchive oa,
            Map<Long, Integer> sessions) throws IOException {
        HashMap<Long, Integer> sessSnap = new HashMap<Long, Integer>(sessions);
        oa.writeInt(sessSnap.size(), "count");
        for (Entry<Long, Integer> entry : sessSnap.entrySet()) {
            oa.writeLong(entry.getKey().longValue(), "id");
            oa.writeInt(entry.getValue().intValue(), "timeout");
        }
        dt.serialize(oa, "tree");
    }

// DataTree
 public void serialize(OutputArchive oa, String tag) throws IOException {
        scount = 0;
        serializeList(longKeyMap, oa);
        serializeNode(oa, new StringBuilder(""));
        // / marks end of stream
        // we need to check if clear had been called in between the snapshot.
        if (root != null) {
            oa.writeString("/", "path");
        }
    }
 

涉及到的几个接口和类



反序列化即把快照文件反序列化成DataTree的过程和序列化的过程正好相反，值得注意的是，反序列化时，找的是最新的，可用的snapshot文件

1. findNValidSnapshots找100个以内的快照文件，并且按照zxid从大到小排列，保证最新的快照最先被处理

2. 如果最新的快照被成功处理，就返回，否则找第二新的快照，直到结束

// FileSnap
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");
return dt.lastProcessedZxid;
}

private List<File> findNValidSnapshots(int n) throws IOException {
        List<File> files = Util.sortDataDir(snapDir.listFiles(),"snapshot", false);
        int count = 0;
        List<File> list = new ArrayList<File>();
        for (File f : files) {
            // we should catch the exceptions
            // from the valid snapshot and continue
            // until we find a valid one
            try {
                if (Util.isValidSnapshot(f)) {
                    list.add(f);
                    count++;
                    if (count == n) {
                        break;
                    }
                }
            } catch (IOException e) {
                LOG.info("invalid snapshot " + f, e);
            }
        }
        return list;
    }

日志文件

ZooKeeper将事务类分为两部分，TxnHeader表示事务头，包含了事务的基本信息。xxxTxn类表示具体类型的事务，包含了事务对应的操作路径和数据。

先来看看ZooKeeper中的事务基础类定义Txn，它只包含了type属性和data属性，实现了Jute Record接口，处理自身的序列化操作。具体的增删改操作事务各自定义了单独的类，都实现了Record接口。单独的事务类和Txn类可以根据type和data互相转化。

public class Txn implements Record {
private int type;
private byte[] data;
  public void serialize(OutputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(this,tag);
    a_.writeInt(type,"type");
    a_.writeBuffer(data,"data");
    a_.endRecord(this,tag);
  }
  public void deserialize(InputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(tag);
    type=a_.readInt("type");
    data=a_.readBuffer("data");
    a_.endRecord(tag);
}

public class CreateTxn implements Record {
  private String path;
  private byte[] data;
  private java.util.List<org.apache.zookeeper.data.ACL> acl;
  private boolean ephemeral;
  private int parentCVersion;
}

public class DeleteTxn implements Record {
  private String path;
}
}

public class SetDataTxn implements Record {
  private String path;
  private byte[] data;
  private int version;
}


单独事务类和Txn类互相转换的示例如下
for (Txn subtxn : txns) {
ByteBuffer bb = ByteBuffer.wrap(subtxn.getData());
Record record = null;
switch (subtxn.getType()) {
case OpCode.create:
record = new CreateTxn();
break;
case OpCode.delete:
record = new DeleteTxn();
break;
case OpCode.setData:
record = new SetDataTxn();
break;
case OpCode.error:
record = new ErrorTxn();
post_failed = true;
break;
case OpCode.check:
record = new CheckVersionTxn();
break;
default:
throw new IOException("Invalid type of op: " + subtxn.getType());
}

ByteBufferInputStream.byteBuffer2Record(bb, record);
}


TxnHeader类定义了事务的基本信息，通过type可以确定具体的事务类型

public class TxnHeader implements Record {
  private long clientId;
  private int cxid;
  private long zxid;
  private long time;
  private int type;
  public TxnHeader() {
  }
  public void serialize(OutputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(this,tag);
    a_.writeLong(clientId,"clientId");
    a_.writeInt(cxid,"cxid");
    a_.writeLong(zxid,"zxid");
    a_.writeLong(time,"time");
    a_.writeInt(type,"type");
    a_.endRecord(this,tag);
  }
  public void deserialize(InputArchive a_, String tag) throws java.io.IOException {
    a_.startRecord(tag);
    clientId=a_.readLong("clientId");
    cxid=a_.readInt("cxid");
    zxid=a_.readLong("zxid");
    time=a_.readLong("time");
    type=a_.readInt("type");
    a_.endRecord(tag);
}
TxnHeader和具体事务类的交互例子如下：

switch (header.getType()) {
case OpCode.create:
CreateTxn createTxn = (CreateTxn) txn;
rc.path = createTxn.getPath();
createNode(
createTxn.getPath(),
createTxn.getData(),
createTxn.getAcl(),
createTxn.getEphemeral() ? header.getClientId() : 0,
createTxn.getParentCVersion(),
header.getZxid(), header.getTime());
break;
case OpCode.delete:
DeleteTxn deleteTxn = (DeleteTxn) txn;
rc.path = deleteTxn.getPath();
deleteNode(deleteTxn.getPath(), header.getZxid());
break;
case OpCode.setData:
SetDataTxn setDataTxn = (SetDataTxn) txn;
rc.path = setDataTxn.getPath();
rc.stat = setData(setDataTxn.getPath(), setDataTxn
.getData(), setDataTxn.getVersion(), header
.getZxid(), header.getTime());
break;

ZooKeeper的日志文件接口是TxnLog接口，它提供了对事务日志的操作。

public interface TxnLog {

void rollLog() throws IOException;

boolean append(TxnHeader hdr, Record r) throws IOException;

TxnIterator read(long zxid) throws IOException;

long getLastLoggedZxid() throws IOException;

boolean truncate(long zxid) throws IOException;

long getDbId() throws IOException;

void commit() throws IOException;

void close() throws IOException;
}

public interface TxnIterator {
        TxnHeader getHeader();
        
        Record getTxn();
  
        boolean next() throws IOException;
 
        void close() throws IOException;
    }

}


TxnLog的默认实现类是FileTxnLog，从它的描述可以看到事务日志文件的格式如下：
FileHeader TxnList ZeroPad

其中FileHeader和数据文件中的FileHeader一样，三要素。
FileHeader: {
magic 4bytes (ZKLG)
version 4bytes
dbid 8bytes
}

TxnList是事务列表，Txn表示单个事务，格式如下，Record表示具体的事务类
checksum Txnlen TxnHeader Record 0x42

来看一下FileTxnLog如何写入一条事务日志
1. append操作负责写入一条事务日志，一条事务日志包含了TxnHeader和Record两部分，这个方法是同步方法，一次只能有一个线程写日志。

2. lastZxidSeen表示当前这个类处理过的最新的事务id，如果要写入的事务id比lastZxidSeen小，记录warn信息

3. 如果文件输出流为空，就新建一个文件输出流，文件名是log.zxid

4. 先写FileHeader文件头

5. padFile(fos)扩大文件大小，在当前位置距离文件尾部还有4KB的时候会扩大文件。 currentSize记录了当前的文件大小

6. 把TxnHeader和Txn序列化到byte数组

7. 计算checksum

8. 先写checksum，再写事务的byte数组，最后写入0x42表示end of record， 写入一条事务结束

public synchronized boolean append(TxnHeader hdr, Record txn)
throws IOException
{
if (hdr != null) {
if (hdr.getZxid() <= lastZxidSeen) {
LOG.warn("Current zxid " + hdr.getZxid()
+ " is <= " + lastZxidSeen + " for "
+ hdr.getType());
}
if (logStream==null) {
if(LOG.isInfoEnabled()){
LOG.info("Creating new log file: log." +
Long.toHexString(hdr.getZxid()));
}

logFileWrite = new File(logDir, ("log." +
Long.toHexString(hdr.getZxid())));
fos = new FileOutputStream(logFileWrite);
logStream=new BufferedOutputStream(fos);
oa = BinaryOutputArchive.getArchive(logStream);
FileHeader fhdr = new FileHeader(TXNLOG_MAGIC,VERSION, dbId);
fhdr.serialize(oa, "fileheader");
// Make sure that the magic number is written before padding.
logStream.flush();
currentSize = fos.getChannel().position();
streamsToFlush.add(fos);
}
padFile(fos);
byte[] buf = Util.marshallTxnEntry(hdr, txn);
if (buf == null || buf.length == 0) {
throw new IOException("Faulty serialization for header " +
"and txn");
}
Checksum crc = makeChecksumAlgorithm();
crc.update(buf, 0, buf.length);
oa.writeLong(crc.getValue(), "txnEntryCRC");
Util.writeTxnBytes(oa, buf);

return true;
}
return false;
}

看看如何获取日志文件中最新的事务id
1. 先从日志目录下，zxid最大的日志文件名获取zxid

2. 然后根据这个zxid获得从这个事务id开始的事务链TxnIterator。遍历这个事务链表，最后的事务就是最新的事务

public long getLastLoggedZxid() {
File[] files = getLogFiles(logDir.listFiles(), 0);
long maxLog=files.length>0?
Util.getZxidFromName(files[files.length-1].getName(),"log"):-1;

// if a log file is more recent we must scan it to find
// the highest zxid
long zxid = maxLog;
TxnIterator itr = null;
try {
FileTxnLog txn = new FileTxnLog(logDir);
itr = txn.read(maxLog);
while (true) {
if(!itr.next())
break;
TxnHeader hdr = itr.getHeader();
zxid = hdr.getZxid();
}
} catch (IOException e) {
LOG.warn("Unexpected exception", e);
} finally {
close(itr);
}
return zxid;
}

FileTxnLog的commit方法保证日志文件的持久化，flush方法是将文件文件内容写到页缓存。如果强制更新到硬盘，调用FileChannel的force方法强制从页缓存刷新到硬盘，并且记录写硬盘的时间，如果超过阀值就记录warn信息
public synchronized void commit() throws IOException {
if (logStream != null) {
logStream.flush();
}
for (FileOutputStream log : streamsToFlush) {
log.flush();
if (forceSync) {
long startSyncNS = System.nanoTime();

log.getChannel().force(false);

long syncElapsedMS =
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startSyncNS);
if (syncElapsedMS > fsyncWarningThresholdMS) {
LOG.warn("fsync-ing the write ahead log in "
+ Thread.currentThread().getName()
+ " took " + syncElapsedMS
+ "ms which will adversely effect operation latency. "
+ "See the ZooKeeper troubleshooting guide");
}
}
}
while (streamsToFlush.size() > 1) {
streamsToFlush.removeFirst().close();
}
}
检查点和数据恢复
这里的检查点就是zxid，可以根据zxid找到对应的事务日志文件，然后在最新的快照文件上进行回放进行数据的恢复。

1. Snapshot.deserialize方法会把最新的快照文件反序列化到DataTree对象和Session中去

2. 快照文件中的最大的zxid作为数据文件目前最大的zxid

3. 用这个最大的zxid + 1去事务日志文件中找事务日志

4. 如果找到了正确的事务日志，使用processTransaction方法进行事务日志的回放

5. PlayBackListener接口提供了在回放时的钩子

// FileTxnSnapshot
public long restore(DataTree dt, Map<Long, Integer> sessions,
PlayBackListener listener) throws IOException {
snapLog.deserialize(dt, sessions);
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;
}

 public void processTransaction(TxnHeader hdr,DataTree dt,
            Map<Long, Integer> sessions, Record txn)
        throws KeeperException.NoNodeException {
        ProcessTxnResult rc;
        switch (hdr.getType()) {
        case OpCode.createSession:
            sessions.put(hdr.getClientId(),
                    ((CreateSessionTxn) txn).getTimeOut());
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG,ZooTrace.SESSION_TRACE_MASK,
                        "playLog --- create session in log: 0x"
                                + Long.toHexString(hdr.getClientId())
                                + " with timeout: "
                                + ((CreateSessionTxn) txn).getTimeOut());
            }
            // give dataTree a chance to sync its lastProcessedZxid
            rc = dt.processTxn(hdr, txn);
            break;
        case OpCode.closeSession:
            sessions.remove(hdr.getClientId());
            if (LOG.isTraceEnabled()) {
                ZooTrace.logTraceMessage(LOG,ZooTrace.SESSION_TRACE_MASK,
                        "playLog --- close session in log: 0x"
                                + Long.toHexString(hdr.getClientId()));
            }
            rc = dt.processTxn(hdr, txn);
            break;
        default:
            rc = dt.processTxn(hdr, txn);
        }

        /**
         * Snapshots are lazily created. So when a snapshot is in progress,
         * there is a chance for later transactions to make into the
         * snapshot. Then when the snapshot is restored, NONODE/NODEEXISTS
         * errors could occur. It should be safe to ignore these.
         */
        if (rc.err != Code.OK.intValue()) {
            LOG.debug("Ignoring processTxn failure hdr:" + hdr.getType()
                    + ", error: " + rc.err + ", path: " + rc.path);
        }
    }

 public interface PlayBackListener {
        void onTxnLoaded(TxnHeader hdr, Record rec);
    }