【杂谈】Kafka的日志段为什么不用内存映射？

package org.apache.kafka.storage.internals.log

...public class LogSegment implementsCloseable {

    ...private finalFileRecords log;

    .../*** Append the given messages starting with the given offset. Add

     * an entry to the index if needed.

     *

     * It is assumed this method is being called from within a lock, it is not thread-safe otherwise.

     *

     *@paramlargestOffset The last offset in the message set

     *@paramlargestTimestampMs The largest timestamp in the message set.

     *@paramshallowOffsetOfMaxTimestamp The last offset of earliest batch with max timestamp in the messages to append.

     *@paramrecords The log entries to append.

     *@throwsLogSegmentOffsetOverflowException if the largest offset causes index offset overflow*/
    public void append(longlargestOffset,longlargestTimestampMs,longshallowOffsetOfMaxTimestamp,

                       MemoryRecords records)throwsIOException {if (records.sizeInBytes() > 0) {

            LOGGER.trace("Inserting {} bytes at end offset {} at position {} with largest timestamp {} at offset {}",

                records.sizeInBytes(), largestOffset, log.sizeInBytes(), largestTimestampMs, shallowOffsetOfMaxTimestamp);int physicalPosition =log.sizeInBytes();if (physicalPosition == 0)

                rollingBasedTimestamp=OptionalLong.of(largestTimestampMs);



            ensureOffsetInRange(largestOffset);//append the messages
            long appendedBytes =log.append(records);

            LOGGER.trace("Appended {} to {} at end offset {}", appendedBytes, log.file(), largestOffset);//Update the in memory max timestamp and corresponding offset.
            if (largestTimestampMs >maxTimestampSoFar()) {

                maxTimestampAndOffsetSoFar= newTimestampOffset(largestTimestampMs, shallowOffsetOfMaxTimestamp);

            }//append an entry to the index (if needed)
            // 稀疏索引，有一定的间隔。可以减少索引量
            if (bytesSinceLastIndexEntry >indexIntervalBytes) {

                offsetIndex().append(largestOffset, physicalPosition);

                timeIndex().maybeAppend(maxTimestampSoFar(), shallowOffsetOfMaxTimestampSoFar());

                bytesSinceLastIndexEntry= 0;

            }

            bytesSinceLastIndexEntry+=records.sizeInBytes();

        }

    }



   ...

}

packageorg.apache.kafka.common.record;



...public class FileRecords extends AbstractRecords implementsCloseable {

     ...private finalFileChannel channel;

     ....public int append(MemoryRecords records) throwsIOException {if (records.sizeInBytes() > Integer.MAX_VALUE -size.get())throw new IllegalArgumentException("Append of size " + records.sizeInBytes() +
                    " bytes is too large for segment with current file position at " +size.get());int written =records.writeFullyTo(channel);

        size.getAndAdd(written);returnwritten;

    }

  

    ...





}

packageorg.apache.kafka.common.record;

....public class MemoryRecords extendsAbstractRecords {

      ...private finalByteBuffer buffer;

      .../*** Write all records to the given channel (including partial records).

     *@paramchannel The channel to write to

     *@returnThe number of bytes written

     *@throwsIOException For any IO errors writing to the channel*/
    public int writeFullyTo(GatheringByteChannel channel) throwsIOException {

        buffer.mark();int written = 0;while (written <sizeInBytes())

            written+=channel.write(buffer);

        buffer.reset();returnwritten;

    }



    ....

}