RocketMQ学习六-消息存储

/**
 * 此类定义要实现的关联接口，从而允许第三方供应商使用自定义消息存储
 */
public interface MessageStore {

    /**
     * 加载以前存储的消息
     */
    boolean load();

    /**
     * 启动此消息存储
     */
    void start() throws Exception;

    /**
     * 关闭消息存储
     */
    void shutdown();

    /**
     * 销毁消息存储，通常，调用之后所有持久化文件将被移除
     */
    void destroy();

    /** 
     *  将消息以异步方式存储到存储器中，处理器可以处理下一个请求，而不是在结果完成后等待结果，以异步方式通知客户端
     * @param msg MessageInstance to store
     * @return a CompletableFuture for the result of store operation
     */
    default CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
        return CompletableFuture.completedFuture(putMessage(msg));
    }

    /**
     * 将批量消息以异步的方式存储到存储器中
     * @param messageExtBatch the message batch
     * @return a CompletableFuture for the result of store operation
     */
    default CompletableFuture<PutMessageResult> asyncPutMessages(final MessageExtBatch messageExtBatch) {
        return CompletableFuture.completedFuture(putMessages(messageExtBatch));
    }

    /**
     * 将消息存储到存储器中
     * @param msg Message instance to store
     * @return result of store operation.
     */
    PutMessageResult putMessage(final MessageExtBrokerInner msg);

    /**
     * 将批消息存储到存储器中
     */
    PutMessageResult putMessages(final MessageExtBatch messageExtBatch);

    /**
     * 从给定的偏移量开始最多查询queueId所属主题的maxMsgNumsmessages。 将使用提供的消息过滤器进一步筛选结果消息。
     */
    GetMessageResult getMessage(final String group, final String topic, final int queueId,
        final long offset, final int maxMsgNums, final MessageFilter messageFilter);

    /**
     * Get maximum offset of the topic queue.
     * 获取主题队列的最大偏移量
     */
    long getMaxOffsetInQueue(final String topic, final int queueId);

    /**
     * Get the minimum offset of the topic queue.
     * 获取主题队列的最小偏移量
     */
    long getMinOffsetInQueue(final String topic, final int queueId);

    /**
     * Get the offset of the message in the commit log, which is also known as physical offset.
     * 获取commitLog中消息的偏移量，也称为物理偏移量。
     */
    long getCommitLogOffsetInQueue(final String topic, final int queueId, final long consumeQueueOffset);

    /**
     * Look up the physical offset of the message whose store timestamp is as specified.
     * 查找其存储时间戳如指定的消息的物理偏移量
     */
    long getOffsetInQueueByTime(final String topic, final int queueId, final long timestamp);

    /**
     * Look up the message by given commit log offset.
     * 查找消息通过commitLog的偏移量
     */
    MessageExt lookMessageByOffset(final long commitLogOffset);

    /**
     * Get one message from the specified commit log offset.
     * 获取一个消息从特定commitLog的偏移中获取
     */
    SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset);

    /**
     * Get one message from the specified commit log offset.
     *  获取一个消息从特定commitLog的offset中获取，带消息大小
     */
    SelectMappedBufferResult selectOneMessageByOffset(final long commitLogOffset, final int msgSize);

    /**
     * 获取这个存储器的运行信息
     */
    String getRunningDataInfo();

    /**
     * 消息存储运行信息，包含各种统计信息
     */
    HashMap<String, String> getRuntimeInfo();

    /**
     * Get the maximum commit log offset.
     * 获取最大提交commitLog的offset
     */
    long getMaxPhyOffset();

    /**
     * Get the minimum commit log offset.
     *获取最小提交commitLog的offset
     */
    long getMinPhyOffset();

    /**
     * Get the store time of the earliest message in the given queue.
     * 获取最早消息在给定队列中的存储时间
     */
    long getEarliestMessageTime(final String topic, final int queueId);

    /**
     * Get the store time of the earliest message in this store.
     * 获取最早消息在给定队列中的存储时间
     * @return timestamp of the earliest message in this store.
     */
    long getEarliestMessageTime();

    /**
     * 获取特定消息的存储时间
     */
    long getMessageStoreTimeStamp(final String topic, final int queueId, final long consumeQueueOffset);

    /**
     * 获取特定队列消息的总数
     */
    long getMessageTotalInQueue(final String topic, final int queueId);

    /**
     * 获取commitLog数据
     */
    SelectMappedBufferResult getCommitLogData(final long offset);

    /**
     * Append data to commit log.
     *  追加数据到commitLog中
     */
    boolean appendToCommitLog(final long startOffset, final byte[] data);

    /**
     * Execute file deletion manually.
     * 手动执行文件删除。
     */
    void executeDeleteFilesManually();

    /**
     * Query messages by given key.
     * 查询消息 通过给定的key
     */
    QueryMessageResult queryMessage(final String topic, final String key, final int maxNum, final long begin,
        final long end);

    /**
     * 更新HA master的地址
     * @param newAddr new address.
     */
    void updateHaMasterAddress(final String newAddr);

    /**
     * Return how much the slave falls behind.
     *  返回slave落后多少
     * @return number of bytes that slave falls behind.
     */
    long slaveFallBehindMuch();

    /**
     * Return the current timestamp of the store.
     * 返回存储的当前时间戳
     * @return current time in milliseconds since 1970-01-01.
     */
    long now();

    /**
     * Clean unused topics.
     * 清理无用的topics
     * @param topics all valid topics.
     * @return number of the topics deleted.
     */
    int cleanUnusedTopic(final Set<String> topics);

    /**
     * Clean expired consume queues.
     * 清理过期的消费者队列
     */
    void cleanExpiredConsumerQueue();

    /**
     * Check if the given message has been swapped out of the memory.
     * 检查给定的消息是否已被换出内存
     */
    boolean checkInDiskByConsumeOffset(final String topic, final int queueId, long consumeOffset);

    /**
     * 获取已存储在提交日志中但尚未调度到使用队列的字节数。
     * @return number of the bytes to dispatch.
     */
    long dispatchBehindBytes();

    /**
     * Flush the message store to persist all data.
     * 刷新消息存储去持久化所有数据
     * @return maximum offset flushed to persistent storage device.
     */
    long flush();

    /**
     * Reset written offset.
     * 重新写入偏移量
     */
    boolean resetWriteOffset(long phyOffset);

    /**
     * Get confirm offset.
     * 获取确认的偏移量
     * @return confirm offset.
     */
    long getConfirmOffset();

    /**
     * Set confirm offset.
     * 设置确认偏移量
     * @param phyOffset confirm offset to set.
     */
    void setConfirmOffset(long phyOffset);

    /**
     * Check if the operation system page cache is busy or not.
     * 检查操作系统页面缓存是否繁忙
     * @return true if the OS page cache is busy; false otherwise.
     */
    boolean isOSPageCacheBusy();

    /**
     * Get lock time in milliseconds of the store by far.
     *  获得锁定时间（以毫秒为单位）。
     * @return lock time in milliseconds.
     */
    long lockTimeMills();

    /**
     * Check if the transient store pool is deficient.
     *  检查临时存储池是否不足
     */
    boolean isTransientStorePoolDeficient();

    /**
     * Get the dispatcher list.
     * 获取分发列表
     * @return list of the dispatcher.
     */
    LinkedList<CommitLogDispatcher> getDispatcherList();

    /**
     * Get consume queue of the topic/queue.
     * 获取消费队列的topic/queue
     */
    ConsumeQueue getConsumeQueue(String topic, int queueId);

    /**
     * Get BrokerStatsManager of the messageStore.
     * 获取broker统计管理信息关于消息存储的
     * @return BrokerStatsManager.
     */
    BrokerStatsManager getBrokerStatsManager();

    /**
     * handle定时消息服务
     * @param brokerRole
     */
    void handleScheduleMessageService(BrokerRole brokerRole);
}

/**
 * 消息存储配置信息
 */
public class MessageStoreConfig {
    //The root directory in which the log data is kept
    //存储路径的根目录
    @ImportantField
    private String storePathRootDir = System.getProperty("user.home") + File.separator + "store";

    //The directory in which the commitlog is kept
    //commitLog的存储路径
    @ImportantField
    private String storePathCommitLog = System.getProperty("user.home") + File.separator + "store"
        + File.separator + "commitlog";

    // CommitLog file size,default is 1G
    //commitLog 文件大小，默认1G
    private int mappedFileSizeCommitLog = 1024 * 1024 * 1024;

    // ConsumeQueue file size,default is 30W
    //消费者队列文件大小，默认30W
    private int mappedFileSizeConsumeQueue = 300000 * ConsumeQueue.CQ_STORE_UNIT_SIZE;

    // enable consume queue ext
    // 是否使用消费者扩展队列 默认false
    private boolean enableConsumeQueueExt = false;

    // ConsumeQueue extend file size, 48M
    //消费者队列 扩展文件大小 48M
    private int mappedFileSizeConsumeQueueExt = 48 * 1024 * 1024;

    // Bit count of filter bit map.
    // this will be set by pipe of calculate filter bit map.
    //位图大小 消费者扩展队列 64
    private int bitMapLengthConsumeQueueExt = 64;

    // CommitLog flush interval
    // flush data to disk
    //commitLog 刷新间隔 500 刷新数据到disk
    @ImportantField
    private int flushIntervalCommitLog = 500;

    // Only used if TransientStorePool enabled
    // flush data to FileChannel
    //刷新数据到fileChannel
    @ImportantField
    private int commitIntervalCommitLog = 200;

    /**
     * introduced since 4.0.x. Determine whether to use mutex reentrantLock when putting message.<br/>
     * By default it is set to false indicating using spin lock when putting message.
     * 从4.0.x开始引入
     * 确定是否在放置消息时使用互斥的reentrantLock。
     * 默认情况下，将其设置为false表示在放置消息时使用自旋锁。
     */
    private boolean useReentrantLockWhenPutMessage = false;

    // Whether schedule flush,default is real-time
    //是否定时刷新，默认是实时的
    @ImportantField
    private boolean flushCommitLogTimed = false;

    //消费队列刷新间隔
    // ConsumeQueue flush interval
    private int flushIntervalConsumeQueue = 1000;

    // Resource reclaim interval
    //资源重新声明 间隔 10000
    private int cleanResourceInterval = 10000;

    //commitLog 移除间隔 100
    // CommitLog removal interval
    private int deleteCommitLogFilesInterval = 100;

    //消费者队列移除 间隔100
    // ConsumeQueue removal interval
    private int deleteConsumeQueueFilesInterval = 100;

    //强制销毁 mapedFile 间隔 1000*120
    private int destroyMapedFileIntervalForcibly = 1000 * 120;

    //重新删除 挂起文件 间隔 1000*120
    private int redeleteHangedFileInterval = 1000 * 120;

    //当删除的时 默认时间凌晨4点
    // When to delete,default is at 4 am
    @ImportantField
    private String deleteWhen = "04";

    private int diskMaxUsedSpaceRatio = 75;

    //保存文件的时间72小时 也即3天 3天后进行删除
    // The number of hours to keep a log file before deleting it (in hours)
    @ImportantField
    private int fileReservedTime = 72;

    // Flow control for ConsumeQueue
    // 放置消息索引最大高水位 600000
    private int putMsgIndexHightWater = 600000;

    //默认最大的消息 4M
    // The maximum size of message,default is 4M
    private int maxMessageSize = 1024 * 1024 * 4;


    // Whether check the CRC32 of the records consumed.
    // This ensures no on-the-wire or on-disk corruption to the messages occurred.
    // This check adds some overhead,so it may be disabled in cases seeking extreme performance.
    //是否检查消耗的记录的CRC32
    //这样可确保不会对消息进行任何在线或磁盘损坏。
    //此检查会增加一些开销，因此在寻求极端性能的情况下可能会被禁用。
    private boolean checkCRCOnRecover = true;

    //刷新CommitLog时要刷新多少页 4k
    // How many pages are to be flushed when flush CommitLog
    private int flushCommitLogLeastPages = 4;


    // How many pages are to be committed when commit data to file
    //将数据提交到文件时要提交多少页 4页
    private int commitCommitLogLeastPages = 4;


    // Flush page size when the disk in warming state
    //磁盘处于预热状态时刷新页面大小 1024/4*16
    private int flushLeastPagesWhenWarmMapedFile = 1024 / 4 * 16;

    // How many pages are to be flushed when flush ConsumeQueue
    //刷新ConsumeQueue时要刷新多少页 2页
    private int flushConsumeQueueLeastPages = 2;

    private int flushCommitLogThoroughInterval = 1000 * 10;
    private int commitCommitLogThoroughInterval = 200;
    private int flushConsumeQueueThoroughInterval = 1000 * 60;

    //消息在内存中的最大传输字节
    @ImportantField
    private int maxTransferBytesOnMessageInMemory = 1024 * 256;

    //内存中消息的最大传输计数
    @ImportantField
    private int maxTransferCountOnMessageInMemory = 32;

    //磁盘中消息的最大传输字节数
    @ImportantField
    private int maxTransferBytesOnMessageInDisk = 1024 * 64;

    //磁盘中消息的最大传输计数
    @ImportantField
    private int maxTransferCountOnMessageInDisk = 8;

    @ImportantField
    private int accessMessageInMemoryMaxRatio = 40;
    @ImportantField
    private boolean messageIndexEnable = true;
    private int maxHashSlotNum = 5000000;
    private int maxIndexNum = 5000000 * 4;
    private int maxMsgsNumBatch = 64;
    @ImportantField
    private boolean messageIndexSafe = false;
    private int haListenPort = 10912;
    private int haSendHeartbeatInterval = 1000 * 5;
    private int haHousekeepingInterval = 1000 * 20;
    private int haTransferBatchSize = 1024 * 32;
    @ImportantField
    private String haMasterAddress = null;
    private int haSlaveFallbehindMax = 1024 * 1024 * 256;

    //broker的角色 异步master
    @ImportantField
    private BrokerRole brokerRole = BrokerRole.ASYNC_MASTER;

    //刷盘的方式 异步
    @ImportantField
    private FlushDiskType flushDiskType = FlushDiskType.ASYNC_FLUSH;

    //异步刷新的超时时间 1000*5
    private int syncFlushTimeout = 1000 * 5;

    //消息延迟的等级 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
    private long flushDelayOffsetInterval = 1000 * 10;

    //强制清理文件 true
    @ImportantField
    private boolean cleanFileForciblyEnable = true;
    private boolean warmMapedFileEnable = false;
    private boolean offsetCheckInSlave = false;
    private boolean debugLockEnable = false;
    private boolean duplicationEnable = false;
    private boolean diskFallRecorded = true;
    private long osPageCacheBusyTimeOutMills = 1000;
    private int defaultQueryMaxNum = 32;

    //临时存储池启用  也即堆外内存存储 默认不开启
    @ImportantField
    private boolean transientStorePoolEnable = false;
    private int transientStorePoolSize = 5;
    private boolean fastFailIfNoBufferInStorePool = false;

    private boolean enableDLegerCommitLog = false;
    private String dLegerGroup;
    private String dLegerPeers;
    private String dLegerSelfId;

/**
 * 消息
 */
public class Message implements Serializable {
    //主题
    private String topic;
    //标识
    private int flag;
    //配置
    private Map<String, String> properties;
    //消息体
    private byte[] body;
    //事务id
    private String transactionId;
  }

/**
 * 消息的扩展
 */
public class MessageExt extends Message {
    //broker名称
    private String brokerName;

    //队列id
    private int queueId;

    //存储大小
    private int storeSize;

    //队列偏移量
    private long queueOffset;

    //系统标识
    private int sysFlag;
    //产生的时间戳
    private long bornTimestamp;
    //产生的host
    private SocketAddress bornHost;

    //存储时间戳
    private long storeTimestamp;
    //存储host
    private SocketAddress storeHost;
    //消息id
    private String msgId;
    //提交日志偏移量
    private long commitLogOffset;
    private int bodyCRC;
    //重新消费事件
    private int reconsumeTimes;

    //准备事务偏移量
    private long preparedTransactionOffset;
}

public class MessageExtBrokerInner extends MessageExt {
    //配置字符、tags编码
    private String propertiesString;
    private long tagsCode;
}

//放入消息
    @Override
    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
        //检查存储状态：是否关闭、是否可写、是否是salve、是否存储OSPageCacheBusy
        //检查完之后，返回ok状态码
        PutMessageStatus checkStoreStatus = this.checkStoreStatus();
        //判断状态码
        if (checkStoreStatus != PutMessageStatus.PUT_OK) {
            return new PutMessageResult(checkStoreStatus, null);
        }
        //检查消息状态：配置字符长度、消息长度，检查完之后，返回ok的状态码
        PutMessageStatus msgCheckStatus = this.checkMessage(msg);
        if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
            return new PutMessageResult(msgCheckStatus, null);
        }
        //获取系统时间
        long beginTime = this.getSystemClock().now();
        //在提交日志中放入消息
        PutMessageResult result = this.commitLog.putMessage(msg);
        //统计时间
        long elapsedTime = this.getSystemClock().now() - beginTime;
        if (elapsedTime > 500) {
            log.warn("not in lock elapsed time(ms)={}, bodyLength={}", elapsedTime, msg.getBody().length);
        }

        this.storeStatsService.setPutMessageEntireTimeMax(elapsedTime);

        if (null == result || !result.isOk()) {
            this.storeStatsService.getPutMessageFailedTimes().incrementAndGet();
        }

    return result;
}

//存放message
public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
    // Set the storage time
    //设置存储时间戳
    msg.setStoreTimestamp(System.currentTimeMillis());
    // Set the message body BODY CRC (consider the most appropriate setting
    // on the client)
    //设置消息体，使用crc32
    msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    // Back to Results
    AppendMessageResult result = null;
    //获取消息存储统计服务
    StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

    //获取topic、queueId、事务值
    String topic = msg.getTopic();
    int queueId = msg.getQueueId();

    final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    //事务类型
    if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
        || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
        // Delay Delivery
        //延迟发布
        if (msg.getDelayTimeLevel() > 0) {
            if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
            }

            topic = ScheduleMessageService.SCHEDULE_TOPIC;
            queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

            // Backup real topic, queueId
            // backup真实的topic、queueId
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
            MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
            msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

            msg.setTopic(topic);
            msg.setQueueId(queueId);
        }
    }

    //获取产生的host
    InetSocketAddress bornSocketAddress = (InetSocketAddress) msg.getBornHost();
    if (bornSocketAddress.getAddress() instanceof Inet6Address) {
        msg.setBornHostV6Flag();
    }

    //获取存储host
    InetSocketAddress storeSocketAddress = (InetSocketAddress) msg.getStoreHost();
    if (storeSocketAddress.getAddress() instanceof Inet6Address) {
        msg.setStoreHostAddressV6Flag();
    }

    long elapsedTimeInLock = 0;

    MappedFile unlockMappedFile = null;
    MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

    putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
    try {
        long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
        this.beginTimeInLock = beginLockTimestamp;

        // Here settings are stored timestamp, in order to ensure an orderly
        // global
        msg.setStoreTimestamp(beginLockTimestamp);

        if (null == mappedFile || mappedFile.isFull()) {
            mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
        }
        if (null == mappedFile) {
            log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
            beginTimeInLock = 0;
            return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
        }

        //追加消息
        result = mappedFile.appendMessage(msg, this.appendMessageCallback);
        //追加消息结果状态 判断状态，并放入信息
        switch (result.getStatus()) {
            case PUT_OK:
                break;
            case END_OF_FILE:
                unlockMappedFile = mappedFile;
                // Create a new file, re-write the message
                mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                if (null == mappedFile) {
                    // XXX: warn and notify me
                    log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                }
                result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                break;
            case MESSAGE_SIZE_EXCEEDED:
            case PROPERTIES_SIZE_EXCEEDED:
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
            case UNKNOWN_ERROR:
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            default:
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
        }

        elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
        beginTimeInLock = 0;
    } finally {
        putMessageLock.unlock();
    }

    if (elapsedTimeInLock > 500) {
        log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
    }

    if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
        this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
    }

    PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

    // Statistics
    //统计信息
    storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
    storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
   //解决硬盘刷新操作
    handleDiskFlush(result, putMessageResult, msg);
    //进行HA同步
    handleHA(result, putMessageResult, msg);

    return putMessageResult;
}

//追加消息
public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb) {
    //追加消息 内部
    return appendMessagesInner(msg, cb);
}

//追加消息
    public AppendMessageResult appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb) {
        assert messageExt != null;
        assert cb != null;

        //获取当前的写位置 可以理解成写指针
        int currentPos = this.wrotePosition.get();

        //如果当前的写位置<文件大小，则进行只加操作
        if (currentPos < this.fileSize) {
            //slice操作：通过slice创建的新缓冲区只能操作原始缓冲区中数组剩余的数据，
            // 即索引为调用slice方法时原始缓冲区的position到limit索引之间的数据，
            // 超出这个范围的数据通过slice创建的新缓冲区无法操作到

            //通过slice操作拿到剩余的元素空间，将当前位置设置为byteBuffer的当前位置，对消息扩展进行判断，如果属于消息扩展brokerInner
            //则进行追加操作，同时分为单个、批量的，否则为错误消息直接返回
            //操作完，进行位置的添加操作，同时拿到操作时间，返回
            ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
            byteBuffer.position(currentPos);
            AppendMessageResult result;
            if (messageExt instanceof MessageExtBrokerInner) {
                //执行消息追加操作
                result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBrokerInner) messageExt);
            } else if (messageExt instanceof MessageExtBatch) {
                //执行消息批量追加操作
                result = cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, (MessageExtBatch) messageExt);
            } else {
                return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
            }
            this.wrotePosition.addAndGet(result.getWroteBytes());
            this.storeTimestamp = result.getStoreTimestamp();
            return result;
        }
        log.error("MappedFile.appendMessage return null, wrotePosition: {} fileSize: {}", currentPos, this.fileSize);
        return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
    }

//返回追加结果
 public AppendMessageResult(AppendMessageStatus status) {
        this(status, 0, 0, "", 0, 0, 0);
    }

    //追加消息结果信息 构造函数
    public AppendMessageResult(AppendMessageStatus status, long wroteOffset, int wroteBytes, String msgId,
        long storeTimestamp, long logicsOffset, long pagecacheRT) {
        this.status = status;
        this.wroteOffset = wroteOffset;
        this.wroteBytes = wroteBytes;
        this.msgId = msgId;
        this.storeTimestamp = storeTimestamp;
        this.logicsOffset = logicsOffset;
        this.pagecacheRT = pagecacheRT;
    }

//进行append操作
public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
    final MessageExtBrokerInner msgInner) {
    // STORETIMESTAMP + STOREHOSTADDRESS + OFFSET <br>

    // PHY OFFSET
    //物理偏移量 写入偏移量
    long wroteOffset = fileFromOffset + byteBuffer.position();
    //拿到标识信息
    int sysflag = msgInner.getSysFlag();
    //产生host的长度、存储host长度、申请容量bornHostHolder、申请容量storeHostHolder
    int bornHostLength = (sysflag & MessageSysFlag.BORNHOST_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
    int storeHostLength = (sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
    ByteBuffer bornHostHolder = ByteBuffer.allocate(bornHostLength);
    ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);
    //重置byteBuffer
    this.resetByteBuffer(storeHostHolder, storeHostLength);
    String msgId;
    //创建消息id
    if ((sysflag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
        msgId = MessageDecoder.createMessageId(this.msgIdMemory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
    } else {
        msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, msgInner.getStoreHostBytes(storeHostHolder), wroteOffset);
    }

    // Record ConsumeQueue information
    //记录消息队列信息
    keyBuilder.setLength(0);
    keyBuilder.append(msgInner.getTopic());
    keyBuilder.append('-');
    keyBuilder.append(msgInner.getQueueId());
    String key = keyBuilder.toString();
    //队列偏移量
    Long queueOffset = CommitLog.this.topicQueueTable.get(key);
    if (null == queueOffset) {
        queueOffset = 0L;
        CommitLog.this.topicQueueTable.put(key, queueOffset);
    }

    // Transaction messages that require special handling
    //事务类型
    final int tranType = MessageSysFlag.getTransactionValue(msgInner.getSysFlag());
    switch (tranType) {
        // Prepared and Rollback message is not consumed, will not enter the
        // consumer queuec
        //准备和回滚消息没有被消费，将不会加入到消费者队列中
        case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
        case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            queueOffset = 0L;
            break;
        case MessageSysFlag.TRANSACTION_NOT_TYPE:
        case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
        default:
            break;
    }

    /**
     * Serialize message
     * 序列化消息，进行编码
     */
    final byte[] propertiesData =
        msgInner.getPropertiesString() == null ? null : msgInner.getPropertiesString().getBytes(MessageDecoder.CHARSET_UTF8);

    final int propertiesLength = propertiesData == null ? 0 : propertiesData.length;
    //配置长度，也即消息主题的长度不能大于127
    if (propertiesLength > Short.MAX_VALUE) {
        log.warn("putMessage message properties length too long. length={}", propertiesData.length);
        return new AppendMessageResult(AppendMessageStatus.PROPERTIES_SIZE_EXCEEDED);
    }
    //主题数据
    final byte[] topicData = msgInner.getTopic().getBytes(MessageDecoder.CHARSET_UTF8);
    final int topicLength = topicData.length;

    final int bodyLength = msgInner.getBody() == null ? 0 : msgInner.getBody().length;
    //消息长度
    final int msgLen = calMsgLength(msgInner.getSysFlag(), bodyLength, topicLength, propertiesLength);

    // Exceeds the maximum message
    if (msgLen > this.maxMessageSize) {
        CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLength
            + ", maxMessageSize: " + this.maxMessageSize);
        return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
    }

    // Determines whether there is sufficient free space
    //是否有充足的容量
    if ((msgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
        this.resetByteBuffer(this.msgStoreItemMemory, maxBlank);
        // 1 TOTALSIZE
        this.msgStoreItemMemory.putInt(maxBlank);
        // 2 MAGICCODE
        this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
        // 3 The remaining space may be any value
        // Here the length of the specially set maxBlank
        final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
        byteBuffer.put(this.msgStoreItemMemory.array(), 0, maxBlank);
        return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgId, msgInner.getStoreTimestamp(),
            queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
    }

    // Initialization of storage space
    //初始化存储空间，放入消息的相关属性信息、配置
    this.resetByteBuffer(msgStoreItemMemory, msgLen);
    // 1 TOTALSIZE
    this.msgStoreItemMemory.putInt(msgLen);
    // 2 MAGICCODE
    this.msgStoreItemMemory.putInt(CommitLog.MESSAGE_MAGIC_CODE);
    // 3 BODYCRC
    this.msgStoreItemMemory.putInt(msgInner.getBodyCRC());
    // 4 QUEUEID
    this.msgStoreItemMemory.putInt(msgInner.getQueueId());
    // 5 FLAG
    this.msgStoreItemMemory.putInt(msgInner.getFlag());
    // 6 QUEUEOFFSET
    this.msgStoreItemMemory.putLong(queueOffset);
    // 7 PHYSICALOFFSET
    this.msgStoreItemMemory.putLong(fileFromOffset + byteBuffer.position());
    // 8 SYSFLAG
    this.msgStoreItemMemory.putInt(msgInner.getSysFlag());
    // 9 BORNTIMESTAMP
    this.msgStoreItemMemory.putLong(msgInner.getBornTimestamp());
    // 10 BORNHOST
    this.resetByteBuffer(bornHostHolder, bornHostLength);
    this.msgStoreItemMemory.put(msgInner.getBornHostBytes(bornHostHolder));
    // 11 STORETIMESTAMP
    this.msgStoreItemMemory.putLong(msgInner.getStoreTimestamp());
    // 12 STOREHOSTADDRESS
    this.resetByteBuffer(storeHostHolder, storeHostLength);
    this.msgStoreItemMemory.put(msgInner.getStoreHostBytes(storeHostHolder));
    // 13 RECONSUMETIMES
    this.msgStoreItemMemory.putInt(msgInner.getReconsumeTimes());
    // 14 Prepared Transaction Offset
    this.msgStoreItemMemory.putLong(msgInner.getPreparedTransactionOffset());
    // 15 BODY
    this.msgStoreItemMemory.putInt(bodyLength);
    if (bodyLength > 0)
        this.msgStoreItemMemory.put(msgInner.getBody());
    // 16 TOPIC
    this.msgStoreItemMemory.put((byte) topicLength);
    this.msgStoreItemMemory.put(topicData);
    // 17 PROPERTIES
    this.msgStoreItemMemory.putShort((short) propertiesLength);
    if (propertiesLength > 0)
        this.msgStoreItemMemory.put(propertiesData);

    final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
    // Write messages to the queue buffer
    byteBuffer.put(this.msgStoreItemMemory.array(), 0, msgLen);

    //追加消息结果
    AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, msgLen, msgId,
        msgInner.getStoreTimestamp(), queueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
   //事务类型
    switch (tranType) {
        case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
        case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
            break;
        case MessageSysFlag.TRANSACTION_NOT_TYPE:
        case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
            // The next update ConsumeQueue information
            CommitLog.this.topicQueueTable.put(key, ++queueOffset);
            break;
        default:
            break;
    }
    return result;
}

 //进行追加操作
    public AppendMessageResult doAppend(final long fileFromOffset, final ByteBuffer byteBuffer, final int maxBlank,
        final MessageExtBatch messageExtBatch) {
        //对byteBuffer进行标记,获取物理偏移量，记录消费队列信息（长度、消息主题、消息的队列id）
        byteBuffer.mark();
        //physical offset
        long wroteOffset = fileFromOffset + byteBuffer.position();
        // Record ConsumeQueue information
        keyBuilder.setLength(0);
        keyBuilder.append(messageExtBatch.getTopic());
        keyBuilder.append('-');
        keyBuilder.append(messageExtBatch.getQueueId());
        String key = keyBuilder.toString();
        //拿到队列的偏移量
        Long queueOffset = CommitLog.this.topicQueueTable.get(key);
        if (null == queueOffset) {
            queueOffset = 0L;
            CommitLog.this.topicQueueTable.put(key, queueOffset);
        }
        //记录开始的偏移量
        long beginQueueOffset = queueOffset;
        int totalMsgLen = 0;
        int msgNum = 0;
        msgIdBuilder.setLength(0);
        final long beginTimeMills = CommitLog.this.defaultMessageStore.now();
        ByteBuffer messagesByteBuff = messageExtBatch.getEncodedBuff();

        int sysFlag = messageExtBatch.getSysFlag();
        //存储host长度
        int storeHostLength = (sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 + 4 : 16 + 4;
        ByteBuffer storeHostHolder = ByteBuffer.allocate(storeHostLength);

        this.resetByteBuffer(storeHostHolder, storeHostLength);
        ByteBuffer storeHostBytes = messageExtBatch.getStoreHostBytes(storeHostHolder);
        //对消息byteBuff进行标记
        messagesByteBuff.mark();
        //hasRemaining判断在当前位置postion和限制limit之间是否有元素
        //也即postion<limit是否为true
        while (messagesByteBuff.hasRemaining()) {
            // 1 TOTALSIZE
            //拿到当前的位置，当前的长度、消息体的长度
            final int msgPos = messagesByteBuff.position();
            final int msgLen = messagesByteBuff.getInt();
            final int bodyLen = msgLen - 40; //only for log, just estimate it
            // Exceeds the maximum message
            //如果消息长度大于消息最大长度，进行提示并返回
            if (msgLen > this.maxMessageSize) {
                CommitLog.log.warn("message size exceeded, msg total size: " + msgLen + ", msg body size: " + bodyLen
                    + ", maxMessageSize: " + this.maxMessageSize);
                return new AppendMessageResult(AppendMessageStatus.MESSAGE_SIZE_EXCEEDED);
            }
            //否者叠加长度
            totalMsgLen += msgLen;
            //查看是否有充足的空间
            // Determines whether there is sufficient free space
            if ((totalMsgLen + END_FILE_MIN_BLANK_LENGTH) > maxBlank) {
                //重置byteBuffer、放入最大长度、将commitLog设置为空白魔法code
                //进行reset操作，此时会恢复到mark的位置

                //标记是一个索引，通过Buffer中的mark()方法指定Buffer中一个特定的position，
                //之后可以通过调用reset()方法恢复到这个position。
                this.resetByteBuffer(this.msgStoreItemMemory, 8);
                // 1 TOTALSIZE
                this.msgStoreItemMemory.putInt(maxBlank);
                // 2 MAGICCODE
                this.msgStoreItemMemory.putInt(CommitLog.BLANK_MAGIC_CODE);
                // 3 The remaining space may be any value
                //ignore previous read
                messagesByteBuff.reset();
                // Here the length of the specially set maxBlank
                byteBuffer.reset(); //ignore the previous appended messages
                byteBuffer.put(this.msgStoreItemMemory.array(), 0, 8);
                return new AppendMessageResult(AppendMessageStatus.END_OF_FILE, wroteOffset, maxBlank, msgIdBuilder.toString(), messageExtBatch.getStoreTimestamp(),
                    beginQueueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
            }
            //move to add queue offset and commitlog offset
            //移动以添加队列偏移量和commitlog偏移量
            messagesByteBuff.position(msgPos + 20);
            messagesByteBuff.putLong(queueOffset);
            messagesByteBuff.putLong(wroteOffset + totalMsgLen - msgLen);

            //重绕缓存区，此时会将原来的位置进行标记清除，位置postion值归为0，同时limit不变
            //常用于重新读取缓冲区中数据时使用
            storeHostBytes.rewind();
            String msgId;
            if ((sysFlag & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0) {
                msgId = MessageDecoder.createMessageId(this.msgIdMemory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
            } else {
                msgId = MessageDecoder.createMessageId(this.msgIdV6Memory, storeHostBytes, wroteOffset + totalMsgLen - msgLen);
            }
           //进行append操作
            if (msgIdBuilder.length() > 0) {
                msgIdBuilder.append(',').append(msgId);
            } else {
                msgIdBuilder.append(msgId);
            }
            //对队列偏移量、msgNum、当前位置进行叠加
            queueOffset++;
            msgNum++;
            messagesByteBuff.position(msgPos + msgLen);
        }

        messagesByteBuff.position(0);
        messagesByteBuff.limit(totalMsgLen);
        byteBuffer.put(messagesByteBuff);
        messageExtBatch.setEncodedBuff(null);
        AppendMessageResult result = new AppendMessageResult(AppendMessageStatus.PUT_OK, wroteOffset, totalMsgLen, msgIdBuilder.toString(),
            messageExtBatch.getStoreTimestamp(), beginQueueOffset, CommitLog.this.defaultMessageStore.now() - beginTimeMills);
        result.setMsgNum(msgNum);
        CommitLog.this.topicQueueTable.put(key, queueOffset);

        return result;
    }

    private void resetByteBuffer(final ByteBuffer byteBuffer, final int limit) {
        byteBuffer.flip();
        byteBuffer.limit(limit);
    }

}

 // Statistics
 //统计信息
 storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
 storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
//解决硬盘刷新操作
 handleDiskFlush(result, putMessageResult, msg);
 //进行HA同步
 handleHA(result, putMessageResult, msg);

//执行刷盘操作：刷盘操作分为两种：同步刷盘和异步刷盘
public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
    // Synchronization flush
    //同步刷新操作
    if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
        final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
        //如果处于等待存储消息的状态，则进行存储
        if (messageExt.isWaitStoreMsgOK()) {
            //拿到消息的偏移量和消息
            GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
            //放入刷盘消息的请求
            service.putRequest(request);
            CompletableFuture<PutMessageStatus> flushOkFuture = request.future();
            PutMessageStatus flushStatus = null;
            try {
                //获取刷盘状态
                flushStatus = flushOkFuture.get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
                        TimeUnit.MILLISECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                //flushOK=false;
            }
            //如果刷盘状态ok，则进行wakeup操作，如果不ok，则返回超时状态
            if (flushStatus != PutMessageStatus.PUT_OK) {
                log.error("do groupcommit, wait for flush failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags()
                    + " client address: " + messageExt.getBornHostString());
                putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_DISK_TIMEOUT);
            }
        } else {
            service.wakeup();
        }
    }
    // Asynchronous flush
    //否者进行异步刷盘
    else {
        if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            flushCommitLogService.wakeup();
        } else {
            commitLogService.wakeup();
        }
    }
}

//执行HA主从同步操作
public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
    //broker的角色，是否是master,如果是则执行HA操作
    if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
        //拿到HA服务
        HAService service = this.defaultMessageStore.getHaService();
        //查看是否是等待存储消息状态，如果是则查看slave是否处于等待状态
        if (messageExt.isWaitStoreMsgOK()) {
            // Determine whether to wait
            if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
                GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
                service.putRequest(request);
                service.getWaitNotifyObject().wakeupAll();
                PutMessageStatus replicaStatus = null;
                try {
                    replicaStatus = request.future().get(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout(),
                            TimeUnit.MILLISECONDS);
                } catch (InterruptedException | ExecutionException | TimeoutException e) {
                }
                if (replicaStatus != PutMessageStatus.PUT_OK) {
                    log.error("do sync transfer other node, wait return, but failed, topic: " + messageExt.getTopic() + " tags: "
                        + messageExt.getTags() + " client address: " + messageExt.getBornHostNameString());
                    putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT);
                }
            }
            // Slave problem
            //否者告诉生产者，slave不可用
            else {
                // Tell the producer, slave not available
                putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
            }
        }
    }

}

消息存储的入口是DefaultMessageStore。同时可以看到其实现了MessageStore。同时通过MessageStoreConfig可以看到配置信息。同时可以看到putMessage操作作为消息存储的入口，putMessage->appendMessage->appendMessagesInner->doAppend，进行统计操作Statistics(getSinglePutMessageTopicTimesTotal、getSinglePutMessageTopicSizeTotal)，进行刷盘操作handleDiskFlush，handleHA进行主从同步