public class FileTxnLog implements TxnLog {
? ? volatile BufferedOutputStream logStream = null;
? ? volatile OutputArchive oa;
? ? volatile FileOutputStream fos = null;
? ??
? ? // 追加事務(wù)日志
? ? 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();
? ? ? ? }
?
? ? ? ? // 默認(rèn)logStream為空
? ? ? ? if (logStream==null) {
? ? ? ? ? ?if(LOG.isInfoEnabled()){
? ? ? ? ? ? ? ? LOG.info("Creating new log file: " + Util.makeLogName(hdr.getZxid()));
? ? ? ? ? ?}
?
? ? ? ? ? ? // 以下代碼為創(chuàng)建事務(wù)日志文件
? ? ? ? ? ? // 根據(jù)當(dāng)前事務(wù)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);
? ? ? ? }
? ? ? ? // 預(yù)分配代碼在這里
? ? ? ? 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;
? ? }
}

public class FilePadding {
? ? long padFile(FileChannel fileChannel) throws IOException {
? ? ? ? // 針對新文件而言，newFileSize=64M
? ? ? ? long newFileSize = calculateFileSizeWithPadding(fileChannel.position(), currentSize, preAllocSize);
? ? ? ? if (currentSize != newFileSize) {
? ? ? ? ? ? // 將文件擴(kuò)充到64M，全部用0來填充
? ? ? ? ? ? fileChannel.write((ByteBuffer) fill.position(0), newFileSize - fill.remaining());
? ? ? ? ? ? currentSize = newFileSize;
? ? ? ? }
? ? ? ? return currentSize;
? ? }
? ??
? ? // size計(jì)算
? ? 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
? ? ? ? // 初始時(shí)候position=0，fileSize為0，preAllocSize為系統(tǒng)參數(shù)執(zhí)行，默認(rèn)為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:這里確實(shí)沒看懂...
? ? ? ? ? ? if (position > fileSize) {
? ? ? ? ? ? ? ? fileSize = position + preAllocSize;
? ? ? ? ? ? ? ? fileSize -= fileSize % preAllocSize;
? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? fileSize += preAllocSize;
? ? ? ? ? ? }
? ? ? ? }
?
? ? ? ? return fileSize;
? ? }
}

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

public class SyncRequestProcessor extends ZooKeeperCriticalThread implements RequestProcessor {
?? ?public void run() {
? ? ? ? try {
? ? ? ? ? ? int logCount = 0;
?
? ? ? ? ? ? setRandRoll(r.nextInt(snapCount/2));
? ? ? ? ? ? while (true) {
? ? ? ? ? ? ? ? ...
? ? ? ? ? ? ? ? if (si != null) {
? ? ? ? ? ? ? ? ? ? // 追加事務(wù)日志
? ? ? ? ? ? ? ? ? ? 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!");
? ? }
}