RocketMQ架构05：CommitLog深度剖析 - 顺序写的艺术

$HOME/store/commitlog/
├── 00000000000000000000      # 文件1：0 ~ 1GB
├── 00000000001073741824      # 文件2：1GB ~ 2GB
├── 00000000002147483648      # 文件3：2GB ~ 3GB
└── ...

文件名 = 该文件第一条消息的物理偏移量（全局唯一）

文件大小：固定 1GB (1024 * 1024 * 1024 字节)

public class MappedFile {
    // 内存映射文件
    private MappedByteBuffer mappedByteBuffer;
    private FileChannel fileChannel;

    // 构造时创建 MMAP
    public MappedFile(String fileName, int fileSize) {
        File file = new File(fileName);
        this.fileChannel = new RandomAccessFile(file, "rw").getChannel();

        // 内存映射整个文件
        this.mappedByteBuffer = fileChannel.map(
            FileChannel.MapMode.READ_WRITE,  // 可读可写
            0,                                // 起始位置
            fileSize                          // 映射大小（1GB）
        );
    }

    // 追加消息（顺序写）
    public AppendMessageResult appendMessage(ByteBuffer msgBuffer) {
        // 获取当前写位置
        int currentPos = this.wrotePosition.get();

        // 直接写入映射内存
        this.mappedByteBuffer.position(currentPos);
        this.mappedByteBuffer.put(msgBuffer);

        // 更新写位置
        this.wrotePosition.addAndGet(msgBuffer.remaining());

        return new AppendMessageResult(AppendMessageStatus.PUT_OK);
    }

    // 读取消息（零拷贝）
    public SelectMappedBufferResult selectMappedBuffer(int pos, int size) {
        // 创建共享的 ByteBuffer（零拷贝）
        ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
        byteBuffer.position(pos);
        ByteBuffer byteBufferNew = byteBuffer.slice();
        byteBufferNew.limit(size);

        return new SelectMappedBufferResult(byteBufferNew);
    }
}

class FlushRealTimeService extends ServiceThread {
    // 刷盘间隔
    private static final int FLUSH_INTERVAL_MS = 500;

    @Override
    public void run() {
        while (!this.isStopped()) {
            try {
                // 等待刷盘信号或超时
                this.waitForRunning(FLUSH_INTERVAL_MS);

                // 执行刷盘
                long beginTime = System.currentTimeMillis();
                CommitLog.this.mappedFileQueue.flush(0);

                long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
                if (storeTimestamp > 0) {
                    // 更新 checkpoint
                    StoreCheckpoint.flush(storeTimestamp);
                }

                long eclipseTime = System.currentTimeMillis() - beginTime;
                if (eclipseTime > 500) {
                    log.warn("Flush data to disk costs {} ms", eclipseTime);
                }
            } catch (Exception e) {
                log.error("Flush service exception", e);
            }
        }
    }
}

// 刷盘操作
public boolean flush(int flushLeastPages) {
    boolean result = true;
    MappedFile mappedFile = findMappedFileByOffset(this.flushedWhere);

    if (mappedFile != null) {
        long storeTimestamp = mappedFile.getStoreTimestamp();
        // 调用 MMAP 的 force() 方法
        int offset = mappedFile.flush(flushLeastPages);

        this.flushedWhere = mappedFile.getFileFromOffset() + offset;
        result = true;
    }

    return result;
}

# broker.conf

# 刷盘策略
flushDiskType=ASYNC_FLUSH      # 异步刷盘（默认，推荐）
# flushDiskType=SYNC_FLUSH     # 同步刷盘（高可靠场景）

# 异步刷盘参数
flushIntervalCommitLog=500     # 刷盘间隔（毫秒）
flushCommitLogLeastPages=4     # 至少累积多少页才刷盘

public class MappedFile {

    // 文件预热
    public void warmMappedFile(FlushDiskType type, int pages) {
        ByteBuffer byteBuffer = this.mappedByteBuffer.slice();

        int flush = 0;
        long beginTime = System.currentTimeMillis();

        // 按页访问文件（每页 4KB）
        for (int i = 0, j = 0; i < fileSize; i += OS_PAGE_SIZE, j++) {
            byteBuffer.put(i, (byte) 0);

            // 每写入一定页数，刷盘一次
            if (type == FlushDiskType.SYNC_FLUSH) {
                if ((i / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE) >= pages) {
                    flush = i;
                    mappedByteBuffer.force();
                }
            }

            // 防止影响其他操作
            if (j % 1000 == 0) {
                Thread.sleep(0);  // 让出CPU
            }
        }

        // 最后刷盘
        if (type == FlushDiskType.SYNC_FLUSH) {
            mappedByteBuffer.force();
        }

        // mlock：锁定内存，防止被swap
        if (isWarmMapedFileEnable()) {
            mlock();
        }

        long eclipseTime = System.currentTimeMillis() - beginTime;
        log.info("Warm up file {} costs {} ms", fileName, eclipseTime);
    }

    // 锁定内存（防止被交换到磁盘）
    public void mlock() {
        final long address = ((DirectBuffer) mappedByteBuffer).address();
        Pointer pointer = new Pointer(address);

        // 调用系统 mlock
        int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));

        if (ret != 0) {
            log.warn("mlock {} failed", fileName);
        }
    }
}

public class CommitLog {

    // 消息写入主入口
    public PutMessageResult putMessage(MessageExtBrokerInner msg) {
        // 1. 设置存储时间
        msg.setStoreTimestamp(System.currentTimeMillis());

        // 2. 计算消息 CRC
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));

        // 3. 获取当前 MappedFile
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

        if (mappedFile == null || mappedFile.isFull()) {
            // 创建新文件
            mappedFile = this.mappedFileQueue.getLastMappedFile(0);
        }

        // 4. 追加消息到文件
        AppendMessageResult result = mappedFile.appendMessage(
            msg,
            this.appendMessageCallback
        );

        switch (result.getStatus()) {
            case PUT_OK:
                break;
            case END_OF_FILE:
                // 文件写满，重新获取
                mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                break;
            case MESSAGE_SIZE_EXCEEDED:
            case PROPERTIES_SIZE_EXCEEDED:
                return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
            default:
                return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
        }

        // 5. 处理刷盘
        handleDiskFlush(result, msg);

        // 6. 处理主从同步
        handleHA(result, msg);

        return new PutMessageResult(PutMessageStatus.PUT_OK, result);
    }

    // 刷盘处理
    private void handleDiskFlush(AppendMessageResult result, MessageExt msg) {
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getFlushDiskType()) {
            // 同步刷盘
            final GroupCommitService service = this.groupCommitService;
            if (msg.isWaitStoreMsgOK()) {
                GroupCommitRequest request = new GroupCommitRequest(
                    result.getWroteOffset() + result.getWroteBytes()
                );
                service.putRequest(request);
                // 等待刷盘完成
                boolean flushOK = request.waitForFlush(
                    this.defaultMessageStore.getSyncFlushTimeout()
                );
                if (!flushOK) {
                    log.error("Sync flush timeout");
                }
            } else {
                service.wakeup();
            }
        } else {
            // 异步刷盘
            this.flushCommitLogService.wakeup();
        }
    }
}

class DefaultAppendMessageCallback implements AppendMessageCallback {

    @Override
    public AppendMessageResult doAppend(
        final long fileFromOffset,
        final ByteBuffer byteBuffer,
        final int maxBlank,
        final MessageExtBrokerInner msgInner
    ) {
        // 当前文件写位置
        long wroteOffset = fileFromOffset + byteBuffer.position();

        // 计算各部分长度
        final byte[] topicData = msgInner.getTopic().getBytes();
        final int topicLength = topicData.length;

        final byte[] propertiesData = msgInner.getPropertiesString().getBytes();
        final int propertiesLength = propertiesData.length;

        final int msgLen = calMsgLength(
            msgInner.getBody().length,
            topicLength,
            propertiesLength
        );

        // 检查是否超出文件大小
        if ((byteBuffer.position() + msgLen) > maxBlank) {
            // 文件剩余空间不足，填充魔数
            byteBuffer.putInt(maxBlank);
            byteBuffer.putInt(BLANK_MAGIC_CODE);
            return new AppendMessageResult(AppendMessageStatus.END_OF_FILE);
        }

        // 开始序列化
        // 1. TOTALSIZE
        byteBuffer.putInt(msgLen);

        // 2. MAGICCODE
        byteBuffer.putInt(MESSAGE_MAGIC_CODE);

        // 3. BODYCRC
        byteBuffer.putInt(msgInner.getBodyCRC());

        // 4. QUEUEID
        byteBuffer.putInt(msgInner.getQueueId());

        // 5. FLAG
        byteBuffer.putInt(msgInner.getFlag());

        // 6. QUEUEOFFSET
        byteBuffer.putLong(queueOffset);

        // 7. PHYSICALOFFSET
        byteBuffer.putLong(wroteOffset);

        // 8. SYSFLAG
        byteBuffer.putInt(msgInner.getSysFlag());

        // 9. BORNTIMESTAMP
        byteBuffer.putLong(msgInner.getBornTimestamp());

        // 10. BORNHOST
        byteBuffer.put(msgInner.getBornHostBytes());

        // 11. STORETIMESTAMP
        byteBuffer.putLong(msgInner.getStoreTimestamp());

        // 12. STOREHOSTADDRESS
        byteBuffer.put(msgInner.getStoreHostBytes());

        // 13. RECONSUMETIMES
        byteBuffer.putInt(msgInner.getReconsumeTimes());

        // 14. Prepared Transaction Offset
        byteBuffer.putLong(msgInner.getPreparedTransactionOffset());

        // 15. BODY
        byteBuffer.putInt(msgInner.getBody().length);
        byteBuffer.put(msgInner.getBody());

        // 16. TOPIC
        byteBuffer.put((byte) topicLength);
        byteBuffer.put(topicData);

        // 17. PROPERTIES
        byteBuffer.putShort((short) propertiesLength);
        byteBuffer.put(propertiesData);

        return new AppendMessageResult(
            AppendMessageStatus.PUT_OK,
            wroteOffset,
            msgLen,
            msgInner.getMsgId()
        );
    }
}

// 生产者批量发送
List<Message> messages = new ArrayList<>();
for (int i = 0; i < 100; i++) {
    messages.add(new Message("TopicA", ("Msg" + i).getBytes()));
}

// 一次网络调用发送100条
SendResult result = producer.send(messages);

// 性能提升：
// - 减少网络往返次数
// - 一次写入多条消息到 CommitLog

// 异步发送（不阻塞）
producer.send(message, new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {
        // 发送成功回调
    }

    @Override
    public void onException(Throwable e) {
        // 发送失败回调
    }
});

// 业务线程继续执行，不等待发送结果
// 性能提升：提高吞吐量

// 启用堆外内存池（进一步优化性能）
# broker.conf
transientStorePoolEnable=true
transientStorePoolSize=5      # 池大小

// 原理：
// 1. 消息先写入堆外内存（DirectByteBuffer）
// 2. 后台线程批量提交到 PageCache
// 3. 再由 OS 刷盘

// 优势：
// - 减少锁竞争
// - 批量提交减少系统调用

# 查看 CommitLog 写位置
cat /tmp/rocketmq/store/commitlog/00000000000000000000

# 查看刷盘位置
cat /tmp/rocketmq/store/checkpoint

# 监控刷盘延迟
tail -f logs/rocketmqlogs/broker.log | grep "flush"

# 检查磁盘 I/O
iostat -x 1

# 检查是否开启同步刷盘
grep flushDiskType conf/broker.conf

# 解决方案：
# 1. 改为异步刷盘（如果业务允许）
# 2. 使用 SSD
# 3. 启用 TransientStorePool

# 检查 MMAP 映射大小
cat /proc/$(pgrep -f BrokerStartup)/maps | grep commitlog

# 解决方案：
# 1. 调整文件保留时间
# 2. 及时删除过期文件