FileCommitLog extends CommitLog { @Override public CommitLogResult log(LogEntry edit, boolean sync) throws LogNotAvailableException { if (failed) { throw new LogNotAvailableException("file commit log is failed"); } boolean hasListeners = isHasListeners(); if (hasListeners) { sync = true; } if (LOGGER.isLoggable(Level.FINEST)) { LOGGER.log(Level.FINEST, "log {0}", edit); } LogEntryHolderFuture future = new LogEntryHolderFuture(edit, sync); try { queueSize.incrementAndGet(); pendingEntries.incrementAndGet(); writeQueue.put(future); if (!sync) { return new CommitLogResult(future.ack, false , false); } else { if (hasListeners) { future.ack.thenAccept((pos) -> { notifyListeners(pos, edit); }); } return new CommitLogResult(future.ack, false , true); } } catch (InterruptedException err) { Thread.currentThread().interrupt(); throw new LogNotAvailableException(err); } } FileCommitLog(
            Path logDirectory, String tableSpaceName,
            long maxLogFileSize, ExecutorService fsyncThreadPool, StatsLogger statslogger,
            Consumer<FileCommitLog> onClose,
            int maxUnsynchedBatchSize,
            int maxUnsynchedBatchBytes,
            int maxSyncTime,
            boolean requireSync,
            boolean enableO_DIRECT
    ); int getQueueSize(); @Override CommitLogResult log(LogEntry edit, boolean sync); @Override void recovery(LogSequenceNumber snapshotSequenceNumber, BiConsumer<LogSequenceNumber, LogEntry> consumer, boolean fencing); @Override void dropOldLedgers(LogSequenceNumber lastCheckPointSequenceNumber); @Override void startWriting(int expectedReplicaCount); @Override void close(); @Override boolean isClosed(); @Override boolean isFailed(); @Override void clear(); @Override LogSequenceNumber getLastSequenceNumber(); static final String LOGFILEEXTENSION; }
@Test public void testLog() throws Exception { try (FileCommitLogManager manager = new FileCommitLogManager(folder.newFolder().toPath())) { manager.start(); int writeCount = 0; final long _startWrite = System.currentTimeMillis(); try (CommitLog log = manager.createCommitLog("tt", "aa", "nodeid")) { log.startWriting(1); for (int i = 0; i < 10_000; i++) { log.log(LogEntryFactory.beginTransaction(0), false); writeCount++; } } final long _endWrite = System.currentTimeMillis(); AtomicInteger readCount = new AtomicInteger(); try (CommitLog log = manager.createCommitLog("tt", "aa", "nodeid")) { log.recovery(LogSequenceNumber.START_OF_TIME, new BiConsumer<LogSequenceNumber, LogEntry>() { @Override public void accept(LogSequenceNumber t, LogEntry u) { readCount.incrementAndGet(); } }, true); } final long _endRead = System.currentTimeMillis(); assertEquals(writeCount, readCount.get()); System.out.println("Write time: " + (_endWrite - _startWrite) + " ms"); System.out.println("Read time: " + (_endRead - _endWrite) + " ms"); } }