Zookeeper事務日志預分配空間解讀

更新時間：2023年04月04日 09:02:55 作者：恐龍弟旺仔

這篇文章主要介紹了Zookeeper事務日志預分配空間解讀，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教

前言

Zookeeper的通過快照日志和事務日志將內存信息保存下來，記錄下來每次請求的具體信息。

尤其是其事務日志，每次處理事務請求時都需要將其記錄下來。

Zookeeper事務日志的默認存儲方式是磁盤文件，那么Zookeeper的總體性能就受限與磁盤文件的寫入速度。

針對這個瓶頸，Zookeeper做了什么優(yōu)化操作呢，本文我們就一起來了解下。

1.事務日志的預分配

事務日志的添加，我們需要從FileTxnLog.append()方法看起

public class FileTxnLog implements TxnLog {
? ? volatile BufferedOutputStream logStream = null;
? ? volatile OutputArchive oa;
? ? volatile FileOutputStream fos = null;
? ??
? ? // 追加事務日志
? ? public synchronized boolean append(TxnHeader hdr, Record txn)
? ? ? ? throws IOException
? ? {
? ? ? ? if (hdr == null) {
? ? ? ? ? ? return false;
? ? ? ? }
?
? ? ? ? if (hdr.getZxid() <= lastZxidSeen) {
? ? ? ? ? ? LOG.warn("Current zxid " + hdr.getZxid()
? ? ? ? ? ? ? ? ? ? + " is <= " + lastZxidSeen + " for "
? ? ? ? ? ? ? ? ? ? + hdr.getType());
? ? ? ? } else {
? ? ? ? ? ? lastZxidSeen = hdr.getZxid();
? ? ? ? }
?
? ? ? ? // 默認logStream為空
? ? ? ? if (logStream==null) {
? ? ? ? ? ?if(LOG.isInfoEnabled()){
? ? ? ? ? ? ? ? LOG.info("Creating new log file: " + Util.makeLogName(hdr.getZxid()));
? ? ? ? ? ?}
?
? ? ? ? ? ? // 以下代碼為創(chuàng)建事務日志文件
? ? ? ? ? ? // 根據(jù)當前事務ID來創(chuàng)建具體文件名，并寫入文件頭信息
? ? ? ? ? ?logFileWrite = new File(logDir, Util.makeLogName(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();
? ? ? ? ? ?filePadding.setCurrentSize(fos.getChannel().position());
? ? ? ? ? ?streamsToFlush.add(fos);
? ? ? ? }
? ? ? ? // 預分配代碼在這里
? ? ? ? filePadding.padFile(fos.getChannel());
? ? ? ? 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;
? ? }
}

創(chuàng)建FileTxnLog對象時，其logStream屬性為null，所以當?shù)谝淮翁幚硎聞照埱髸r，會先根據(jù)當前事務ID來創(chuàng)建一個文件。

1.1 事務日志預分配

public class FilePadding {
? ? long padFile(FileChannel fileChannel) throws IOException {
? ? ? ? // 針對新文件而言，newFileSize=64M
? ? ? ? long newFileSize = calculateFileSizeWithPadding(fileChannel.position(), currentSize, preAllocSize);
? ? ? ? if (currentSize != newFileSize) {
? ? ? ? ? ? // 將文件擴充到64M，全部用0來填充
? ? ? ? ? ? fileChannel.write((ByteBuffer) fill.position(0), newFileSize - fill.remaining());
? ? ? ? ? ? currentSize = newFileSize;
? ? ? ? }
? ? ? ? return currentSize;
? ? }
? ??
? ? // size計算
? ? public static long calculateFileSizeWithPadding(long position, long fileSize, long preAllocSize) {
? ? ? ? // If preAllocSize is positive and we are within 4KB of the known end of the file calculate a new file size
? ? ? ? // 初始時候position=0，fileSize為0，preAllocSize為系統(tǒng)參數(shù)執(zhí)行，默認為64M
? ? ? ? if (preAllocSize > 0 && position + 4096 >= fileSize) {
? ? ? ? ? ? // If we have written more than we have previously preallocated we need to make sure the new
? ? ? ? ? ? // file size is larger than what we already have
? ? ? ? ? ? // Q:這里確實沒看懂...
? ? ? ? ? ? if (position > fileSize) {
? ? ? ? ? ? ? ? fileSize = position + preAllocSize;
? ? ? ? ? ? ? ? fileSize -= fileSize % preAllocSize;
? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? fileSize += preAllocSize;
? ? ? ? ? ? }
? ? ? ? }
?
? ? ? ? return fileSize;
? ? }
}

預分配的過程比較簡單，就是看下當前文件的剩余空間是否<4096，如果是，則擴容。

Q：

這里有一個不太明白的問題，position > fileSize的場景是怎樣的呢？

2.創(chuàng)建新的事務日志文件時機

通過上述代碼分析我們知道，當logStream=null時，就會創(chuàng)建一個新的事務日志文件，那么logStream對象什么時候為空呢？

搜索代碼，只看到FileTxnLog.rollLog()方法會主動將logStream設置為null

public class FileTxnLog implements TxnLog {
? ? public synchronized void rollLog() throws IOException {
? ? ? ? if (logStream != null) {
? ? ? ? ? ? this.logStream.flush();
? ? ? ? ? ? this.logStream = null;
? ? ? ? ? ? oa = null;
? ? ? ? }
? ? }
}

那么根據(jù)這個線索，我們來搜索下rollLog的調用鏈

SyncRequestProcessor.run() -> ZKDatabase.rollLog() -> FileTxnSnapLog.rollLog() -> FileTxnLog.rollLog()

最終看到是在SyncRequestProcessor.run()方法中發(fā)起調用的，而且只有這一條調用鏈，我們來分析下

2.1 SyncRequestProcessor.run()

public class SyncRequestProcessor extends ZooKeeperCriticalThread implements RequestProcessor {
?? ?public void run() {
? ? ? ? try {
? ? ? ? ? ? int logCount = 0;
?
? ? ? ? ? ? setRandRoll(r.nextInt(snapCount/2));
? ? ? ? ? ? while (true) {
? ? ? ? ? ? ? ? ...
? ? ? ? ? ? ? ? if (si != null) {
? ? ? ? ? ? ? ? ? ? // 追加事務日志
? ? ? ? ? ? ? ? ? ? if (zks.getZKDatabase().append(si)) {
? ? ? ? ? ? ? ? ? ? ? ? logCount++;
? ? ? ? ? ? ? ? ? ? ? ? if (logCount > (snapCount / 2 + randRoll)) {
? ? ? ? ? ? ? ? ? ? ? ? ? ? setRandRoll(r.nextInt(snapCount/2));
? ? ? ? ? ? ? ? ? ? ? ? ? ? // 注意：在這里發(fā)起了rollLog
? ? ? ? ? ? ? ? ? ? ? ? ? ? zks.getZKDatabase().rollLog();
? ? ? ? ? ? ? ? ? ? ? ? ? ? ...
? ? ? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? } else if (toFlush.isEmpty()) {
? ? ? ? ? ? ? ? ? ? ? ? ...
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? ? ? toFlush.add(si);
? ? ? ? ? ? ? ? ? ? if (toFlush.size() > 1000) {
? ? ? ? ? ? ? ? ? ? ? ? flush(toFlush);
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? } catch (Throwable t) {
? ? ? ? ? ? handleException(this.getName(), t);
? ? ? ? ? ? running = false;
? ? ? ? }
? ? ? ? LOG.info("SyncRequestProcessor exited!");
? ? }
}

需要注意下rollLog()方法執(zhí)行的條件，就是logCount > (snapCount / 2 + randRoll)

snapCount是一個系統(tǒng)參數(shù)，System.getProperty("zookeeper.snapCount")，默認值為100000

randRoll是一個隨機值

那么該條件觸發(fā)的時機為：處理的事務請求數(shù)至少要大于50000。

這時就出現(xiàn)了一個筆者無法理解的情況：

通過對事務日志的觀察可以看到其都是64M，而至少處理50000次事務請求后，Zookeeper才會分配一個新的事務日志文件，那么這個snapCount是一個經驗值嘛？

如果事務請求的value信息都很大，那么可能到不了50000次，就會超過64M，理論上應該要創(chuàng)建一個新的文件了，但是貌似并沒有，這個該怎么處理呢？

如果事務請求value信息都很小，那么即使到了50000次，也不會超過64M，那么之前預分配的文件大小就浪費了一部分。