RocketMQ消息队列MQ（Kafka&RabbitMQ）MQ消息系列

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

2017-06-12  本文已影响272人  _呆瓜_

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

1. MappedFile

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

关键字段

关键方法

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.

关键字段

关键方法

(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实现了消息文件队列功能.

关键字段

关键方法

(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字节)

关键方法

(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

上一篇 下一篇

猜你喜欢

热点阅读