RocketMQ源码阅读(四)-消息存储二

RocketMQ的消息存储过程非常复杂, 本文先介绍存储模块中几个重要对象.

1. MappedFile

对MappedByteBuffer的封装, 具有创建文件（使用非堆区内存）, 写入, 提交, 读取, 释放, 关闭等功能, RocketMQ使用该类实现数据从内存到磁盘的持久化.

关键字段

• fileChannel: 该类对应的文件通道.
• mappedByteBuffer: 文件在内存中的映射. 如前文所述RocketMQ使用内存映射的方式来操作文件, 这种方式要比流的方式快很多.
• fileSize: 文件尺寸
• wrotePosition: 当前写到哪一个位置.
• committedPosition: 已经提交(已经持久化到磁盘)的位置.
• flushedPosition: 已经提交(已经持久化到磁盘)的位置.
• writeBuffer: 内存字节缓冲区, RocketMQ提供两种数据落盘的方式: 一种是直接将数据写到映射文件字节缓冲区(mappedByteBuffer), 映射文件字节缓冲区(mappedByteBuffer)flush; 另一种是先写到writeBuffer, 再从内存字节缓冲区(write buffer)提交(commit)到文件通道(fileChannel), 然后文件通道(fileChannel)flush.
• fileFromOffset: fileFromOffset: 映射的起始偏移量, 拿commitlog文件来举例, 下面有很多个文件夹(假设为1KB, 默认是1G大小), 第一个文件名为00000000000000000000, 第二个文件名为00000000000000001024, 那么第一个文件的fileFromOffset就是0, 第二个文件的fileFromOffset就是1024

关键方法

• appendMessage: 插入消息到MappedFile, 并返回插入结果.
• selectMappedBuffer: 返回指定位置的内存映射, 用于读取数据.
  (1) appendMessage
  源代码如下:

public AppendMessageResult appendMessage(final MessageExtBrokerInner msg, final AppendMessageCallback cb) {
    assert msg != null;
    assert cb != null;

    int currentPos = this.wrotePosition.get();  //获取当前写的位置

    if (currentPos < this.fileSize) {   //currentPos小于文件尺寸才能写入
        //获取获取需要写入的字节缓冲区, 之所以会有writeBuffer != null的判断与使用的刷盘服务有关.
        ByteBuffer byteBuffer = writeBuffer != null ? writeBuffer.slice() : this.mappedByteBuffer.slice();
        byteBuffer.position(currentPos);    //设置写入的postion
        AppendMessageResult result =
            cb.doAppend(this.getFileFromOffset(), byteBuffer, this.fileSize - currentPos, msg);  //执行写入
        this.wrotePosition.addAndGet(result.getWroteBytes()); //更新wrotePosition
        this.storeTimestamp = result.getStoreTimestamp();
        return result;
    }
    //返回错误信息
    log.error("MappedFile.appendMessage return null, wrotePosition: " + currentPos + " fileSize: "
        + this.fileSize);
    return new AppendMessageResult(AppendMessageStatus.UNKNOWN_ERROR);
}

可以看到MappedFile调用AppendMessageCallback来执行msg到字节缓冲区的写入.事实上整个RocketMQ只有一个类实现了AppendMessageCallback接口, 就是DefaultAppendMessageCallback. doAppend方法的具体实现与消息格式有关, 并且不属于MappedFile的范畴, 后文再分析.
(2) selectMappedBuffer
源代码如下:

//返回从pos到 pos + size的内存映射
public SelectMappedBufferResult selectMappedBuffer(int pos, int size) {
    int readPosition = getReadPosition();   //获取当前有效数据的最大位置
    if ((pos + size) <= readPosition) {    //内存映射的最大位置必须小于readPosition

        if (this.hold()) {    //引用计数
            ByteBuffer byteBuffer = this.mappedByteBuffer.slice();  // 复制一个byteBuffer(与原byteBuffer共享数据, 只是指针位置独立)
            byteBuffer.position(pos);    //设置position
            //获取目标数据
            ByteBuffer byteBufferNew = byteBuffer.slice();
            byteBufferNew.limit(size);
            return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this);
        } else {
            log.warn("matched, but hold failed, request pos: " + pos + ", fileFromOffset: "
                + this.fileFromOffset);
        }
    } else {
        log.warn("selectMappedBuffer request pos invalid, request pos: " + pos + ", size: " + size
            + ", fileFromOffset: " + this.fileFromOffset);
    }

    return null;
}

2. MappedFileQueue

顾名思义, 该类代表了MappedFile组成的队列(由大小相同的多个文件构成). 无论是CommitLog(消息主体以及元数据), 还是ConsumeQueue(逻辑队列), 底层使用的组件都是MappedFileQueue.

关键字段

• storePath: 文件队列的存储路径
• mappedFiles: 存储MappedFile的map
• mappedFileSize: MappedFile的尺寸
• flushedWhere: 已经刷到磁盘的位置
• committedWhere: 已经提交的位置

关键方法

• getLastMappedFile: 获取队列中最后一个MappedFile对象
• findMappedFileByOffset: 根据offset/filesize计算该offset所在那个文件中

(1) getLastMappedFile
源代码如下:

public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
    long createOffset = -1;
    //获取当前Queue中最后一个MappedFile
    MappedFile mappedFileLast = getLastMappedFile();

    //一个文件都不存在时, 计算起始文件的offset
    if (mappedFileLast == null) {
        createOffset = startOffset - (startOffset % this.mappedFileSize);
    }
    //计算需要新创建的文件的offset
    if (mappedFileLast != null && mappedFileLast.isFull()) {
        createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
    }

    //创建新的MappedFile
    if (createOffset != -1 && needCreate) {
        //计算文件名
        String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
        String nextNextFilePath = this.storePath + File.separator
            + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
        MappedFile mappedFile = null;

        if (this.allocateMappedFileService != null) {
            //使用AllocateMappedFileService创建文件主要是更加安全一些, 会将一些并行的操作穿行化
            mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
                nextNextFilePath, this.mappedFileSize);
        } else {
            try {
                mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
            } catch (IOException e) {
                log.error("create mappedFile exception", e);
            }
        }

        //将新创建的文件添加到队列中
        if (mappedFile != null) {
            if (this.mappedFiles.isEmpty()) {
                mappedFile.setFirstCreateInQueue(true);
            }
            this.mappedFiles.add(mappedFile);
        }

        return mappedFile;
    }

    return mappedFileLast;
}

从源码中可见, 只有当文件写满或者找不到文件时, 才会创建新的文件.
(2) findMappedFileByOffset
主要是根据offset寻找对应的MappedFile, 具体源代码不再贴出.
为了理解findMapedFileByOffset, 我们假设每个文件的大小是1024K, 参考以下图示:

如果现在想查找3021在那个文件中, 可以按如下计算:
(3021 - 0)/1024=2 即可知其在队列下标为2的MappedFile中
释义如下: (offset-第一个文件的fileFromeOffset)/mappedFileSize

3. CommitLog

用于存储消息的抽象封装, 内部采用MapedFileQueue实现了消息文件队列功能.

关键字段

• HashMap topicQueueTable: 用于记录某个topic在某个queueId共写入了多少个消息, put一个消息加1.

关键方法

• putMessage: 存储消息.
• getMessage: 读取消息

(1) putMessage
存储消息主要分3步: 查找文件(getLastMapedFile), 写入数据(DefaultAppendMessageCallback), 刷盘(FlushRealTimeService). 最终产生实际存储消息的队列文件如下:
${storePathRootDir}/commitlog/消息队列文件. (消息队列文件名规则如MappedFileQueue).

(2)getMessage(final long offset, final int size)
offset: 绝对偏移量, 可以用其调用findMappedFileByOffset查询MappedFile.
size: 欲查询的数据大小.

4. ConsumeQueue

消费队列的实现, 该消费队列主要存储了消息在CommitLog的位置, 与CommitLog类似, 内部采用MappedFileQueue实现了消息位置文件队列功能.
一个topic和一个queueId对应一个ConsumeQueue.
默认queue存储30W条消息, 每个消息大小为20个字节, 详细如下:
offset(long 8字节) + size(int 4字节) + tagsCode(long 8字节)

关键方法

• putMessagePositionInfo: 消息位置的存储
• getIndexBuffer: 该方法返回从offset之后的字节映射
  (1)putMessagePositionInfo(final long offset, final int size, final long tagsCode,
  final long cqOffset)
  offset: 消息在commitLog中的起始位置
  size: 消息长度
  tagsCode: 消息tag的hash code
  cqOffset: 该消息在topic对应的queue中的下标
  该方法主要实现了消息位置的存储, 并产生消息文件:
  storePathRootDir/consumequeue/{topic}/${queueId}/消息位置队列文件
  消息数(30W)*消息位置固定大小(20字节)=6000000字节
  故每6000000字节一个文件, 文件名依次递增, 前缀不够20位补0, 类似如下:
  00000000000000000000
  00000000000006000000
  00000000000012000000

(2)getIndexBuffer(final long startIndex)
该方法源代码如下:

public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
    int mappedFileSize = this.mappedFileSize;
    long offset = startIndex * CQ_STORE_UNIT_SIZE;
    if (offset >= this.getMinLogicOffset()) {
        MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset);
        if (mappedFile != null) {
            SelectMappedBufferResult result = mappedFile.selectMappedBuffer((int) (offset % mappedFileSize));
            return result;
        }
    }
    return null;
}

startIndex代表了起始偏移量索引.
该方法先根据startIndex找到对应的MappedFile, 再在该MappedFile中找到对应的字节映射.

5. 总结

RocketMQ的消息存储非常复杂, 本文介绍了消息存储中使用到的基础组件类以及一些重要的API. 后文会进一步介绍消息存储的详细流程.

参考资料:
1.http://www.tuicool.com/articles/6FFR7v
2.http://blog.csdn.net/a417930422/article/details/50606732