墨与星辰

主题

消息存储机制 ​

深入理解RocketMQ的消息存储机制，掌握CommitLog、ConsumeQueue、IndexFile等存储文件的实现，理解消息存储优化、刷盘机制等核心功能。

📋 消息存储概览 ​

RocketMQ采用混合型存储结构，将消息存储和索引分离，提高存储性能。

🎯 一、CommitLog ​

1.1 CommitLog结构 ​

CommitLog是消息的主存储文件，所有消息都顺序写入CommitLog。

CommitLog文件结构：

+--------+--------+--------+--------+--------+--------+--------+--------+
|  Magic |  Body Length  |  Body CRC32  |  Queue ID  |  Queue Offset   |
+--------+--------+--------+--------+--------+--------+--------+--------+
|  Physical Offset  |  Sys Flag  |  Born Timestamp  |  Born Host       |
+--------+--------+--------+--------+--------+--------+--------+--------+
|  Store Timestamp  |  Store Host  |  Reconsume Times  |  Transaction   |
+--------+--------+--------+--------+--------+--------+--------+--------+
|  Prepared Transaction Offset  |  Body  |  Topic Length  |  Topic       |
+--------+--------+--------+--------+--------+--------+--------+--------+
|  Properties Length  |  Properties  |
+--------+--------+--------+--------+

1.2 消息写入 ​

DefaultMessageStore.putMessage：

java
public PutMessageResult putMessage(MessageExtBrokerInner msg) {
    PutMessageStatus checkStoreStatus = this.checkStoreStatus();
    if (checkStoreStatus != PutMessageStatus.PUT_OK) {
        return new PutMessageResult(checkStoreStatus, null);
    }
    
    PutMessageStatus msgCheckStatus = this.checkMessage(msg);
    if (msgCheckStatus == PutMessageStatus.MESSAGE_ILLEGAL) {
        return new PutMessageResult(msgCheckStatus, null);
    }
    
    long beginLockTimestamp = this.getSystemClock().now();
    this.putMessageLock.lock();
    try {
        long msgStoreTimestamp = this.commitLog.getMaxOffset();
        msg.setStoreTimestamp(msgStoreTimestamp);
        
        MappedFile unlockMappedFile = null;
        MappedFile mappedFile = this.commitLog.getLastMappedFile();
        
        msg.setQueueId(queueId);
        msg.setQueueOffset(queueOffset);
        
        AppendMessageResult result = mappedFile.appendMessage(msg);
        
        switch (result.getStatus()) {
            case PUT_OK:
                break;
            case END_OF_FILE:
                unlockMappedFile = mappedFile;
                mappedFile = this.commitLog.getLastMappedFile();
                if (null == mappedFile) {
                    return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                }
                result = mappedFile.appendMessage(msg);
                break;
            case MESSAGE_SIZE_EXCEEDED:
            case PROPERTIES_SIZE_EXCEEDED:
                return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
            case UNKNOWN_ERROR:
                return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            default:
                return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
        }
        
        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
        
        this.handleDiskFlush(result, putMessageResult, msg);
        this.handleHA(result, putMessageResult, msg);
        
        return putMessageResult;
    } finally {
        this.putMessageLock.unlock();
    }
}

🎯 二、ConsumeQueue ​

2.1 ConsumeQueue结构 ​

ConsumeQueue是消费队列索引，按Topic和QueueId组织。

ConsumeQueue条目结构：

+--------+--------+--------+--------+
|  CommitLog Offset (8 bytes)       |
+--------+--------+--------+--------+
|  Message Size (4 bytes)          |
+--------+--------+--------+--------+
|  Message Tag HashCode (8 bytes)  |
+--------+--------+--------+--------+

2.2 ConsumeQueue构建 ​

ConsumeQueue构建流程：

🎯 三、IndexFile ​

3.1 IndexFile结构 ​

IndexFile提供消息索引功能，支持按时间查询消息。

IndexFile结构：

+--------+--------+--------+--------+
|  Header (40 bytes)                |
+--------+--------+--------+--------+
|  Slot Table (500万 * 4 bytes)     |
+--------+--------+--------+--------+
|  Index Linked List (2000万 * 20 bytes) |
+--------+--------+--------+--------+

3.2 索引构建 ​

IndexService.buildIndex：

java
public void buildIndex(DispatchRequest req) {
    IndexFile indexFile = retryGetAndCreateIndexFile();
    if (indexFile != null) {
        long endPhyOffset = indexFile.getEndPhyOffset();
        DispatchRequest msg = req;
        String topic = msg.getTopic();
        String keys = msg.getKeys();
        if (msg.getCommitLogOffset() < endPhyOffset) {
            return;
        }
        
        final int tranType = MessageSysFlag.getTransactionValue(
                msg.getSysFlag());
        switch (tranType) {
            case MessageSysFlag.TRANSACTION_NOT_TYPE:
            case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                break;
            case MessageSysFlag.TRANSACTION_PREPARED_TYPE:
            case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                return;
        }
        
        if (req.getUniqKey() != null) {
            indexFile.putKey(topic, req.getUniqKey(), 
                    req.getCommitLogOffset(), req.getStoreTimestamp());
        }
        
        if (keys != null && keys.length() > 0) {
            String[] keyset = keys.split(MessageConst.KEY_SEPARATOR);
            for (int i = 0; i < keyset.length; i++) {
                String key = keyset[i];
                if (key.length() > 0) {
                    indexFile.putKey(topic, key, req.getCommitLogOffset(), 
                            req.getStoreTimestamp());
                }
            }
        }
    } else {
        log.error("build index error, stop building index");
    }
}

🎯 四、刷盘机制 ​

4.1 同步刷盘 ​

同步刷盘：

java
public void handleDiskFlush(AppendMessageResult result, 
        PutMessageResult putMessageResult, MessageExt messageExt) {
    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);
            boolean flushOK = request.waitForFlush(
                    this.defaultMessageStore.getMessageStoreConfig()
                            .getSyncFlushTimeout());
            if (!flushOK) {
                putMessageResult.setPutMessageStatus(
                        PutMessageStatus.FLUSH_DISK_TIMEOUT);
            }
        } else {
            service.wakeup();
        }
    } else {
        this.flushCommitLogService.wakeup();
    }
}

4.2 异步刷盘 ​

异步刷盘：

java
public void handleDiskFlush(AppendMessageResult result, 
        PutMessageResult putMessageResult, MessageExt messageExt) {
    this.flushCommitLogService.wakeup();
}

📖 学习资源 ​

视频教程 ​

书籍推荐 ​

  • 《RocketMQ技术内幕》 - 丁威
  • 《分布式消息中间件实践》 - 倪炜

网站资源 ​

✅ 学习检查清单 ​

  • [ ] 已理解CommitLog实现
  • [ ] 已理解ConsumeQueue实现
  • [ ] 已理解IndexFile实现
  • [ ] 已理解刷盘机制
  • [ ] 已绘制存储架构图
  • [ ] 已总结核心知识点

上一章节：RocketMQ源码深度解析概览
下一章节：消息投递流程
返回目录：源码阅读