RocketMQ源码深度解析一之消息存储

RocketMQ的消息持久化是基于文件系统，而从效率来看文件系统>kv存储>关系型数据库。那么，到底是如何存储的，相信对源码进行解析，将会是我们大大提高对消息存储的认识。

（一）存储层的整体结构 
首先看下结构图 

对于我们业务层来说，都是通过DefaultMessageStore类做为操作入口。RocketMQ下主要有6类文件，分别是三类大文件：Index文件，consumequeue文件，commitlog文件。三类小文件:checkpoint文件，config目录下的配置文件.和abort。 
而对于三类大文件，使用的就是nio的MappedByteBuffer类来提高读写性能。这个类是文件内存映射的相关类，支持随机读和顺序写。在RocketMQ中，被封装成了MappedFile类。 
RocketMQ对于每类大文件，在存储时候分割成了多个固定大小的文件，每个文件名为前面所有的文件大小加1（也就是偏移量）。从而实现对整个大文件的串联拼接。接下来就需要看看MappedFIle这个封装类了。

（二）MappedFile类 
对于 commitlog、 consumequeue、 index 三类大文件进行磁盘读写操作，均是通过 MapedFile 类来完成。这个类相当于MappedByteBuffer的包装类。 
（1）主要成员变量

    //默认页大小为4k
    public static final int OS_PAGE_SIZE = 1024 * 4;
    protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    //JVM中映射的虚拟内存总大小  
    private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);
    //JVM中mmap的数量  
    private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0);
    //当前写文件的位置
    protected final AtomicInteger wrotePosition = new AtomicInteger(0);
    //ADD BY ChenYang
    protected final AtomicInteger committedPosition = new AtomicInteger(0);
    private final AtomicInteger flushedPosition = new AtomicInteger(0);
    //映射文件的大小
    protected int fileSize;
    //映射的fileChannel对象
    protected FileChannel fileChannel;
    /**
     * Message will put to here first, and then reput to FileChannel if writeBuffer is not null.
     */
    protected ByteBuffer writeBuffer = null;
    protected TransientStorePool transientStorePool = null;
    //映射的文件名
    private String fileName;
    //映射的起始偏移量
    private long fileFromOffset;
    //映射的文件
    private File file;
    //映射的内存对象
    private MappedByteBuffer mappedByteBuffer;
    //最后一条消息保存时间  
    private volatile long storeTimestamp = 0;
    //是不是刚刚创建的
    private boolean firstCreateInQueue = false;

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34

（2）文件顺序写操作 
这里提供了两种顺序写的方法。 
第一个是提供给commitlog使用的，传入消息内容，然后CommitLog按照规定的格式构造二进制信息并顺序写，方法是appendMessage(final Object msg, final AppendMessageCallback cb)；这里的msg是MessageExtBrokerInner或者MessageExtBatch，具体的写操作是在回调类的doAppend方法进行。最后再调用appendMessagesInner(final MessageExt messageExt, final AppendMessageCallback cb)；

第二个是appendMessage(final byte[] data)方法

public boolean appendMessage(final byte[] data) {
        //找出当前要的写入位置
        int currentPos = this.wrotePosition.get();
        //如果当前位置加上要写入的数据大小小于等于文件大小，则说明剩余空间足够写入。
        if ((currentPos + data.length) <= this.fileSize) {
            try {
                //则由内存对象 mappedByteBuffer 创建一个指向同一块内存的
                //ByteBuffer 对象，并将内存对象的写入指针指向写入位置；然后将该二进制信
                //息写入该内存对象，同时将 wrotePostion 值增加消息的大小；
                this.fileChannel.position(currentPos);
                this.fileChannel.write(ByteBuffer.wrap(data));
            } catch (Throwable e) {
                log.error("Error occurred when append message to mappedFile.", e);
            }
            this.wrotePosition.addAndGet(data.length);
            return true;
        }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18

（3）消息刷盘操作 
主要的方法是commit(final int flushLeastPages)；

public int commit(final int commitLeastPages) {
        if (writeBuffer == null) {
            //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
            return this.wrotePosition.get();
        }
        if (this.isAbleToCommit(commitLeastPages)) {
            if (this.hold()) {
                commit0(commitLeastPages);
                this.release();
            } else {
                log.warn("in commit, hold failed, commit offset = " + this.committedPosition.get());
            }
        }

        // All dirty data has been committed to FileChannel.
        if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
            this.transientStorePool.returnBuffer(writeBuffer);
            this.writeBuffer = null;
        }

        return this.committedPosition.get();
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22

解析： 
（1）首先判断文件是否已经写满类，即wrotePosition等于fileSize，若写慢则进行刷盘操作 
（2）检测内存中尚未刷盘的消息页数是否大于最小刷盘页数，不够页数也暂时不刷盘。 
（3）MappedFile的父类是ReferenceResource，该父类作用是记录MappedFile中的引用次数，为正表示资源可用，刷盘前加一，然后将wrotePosotion的值赋给committedPosition，再减一。

（4）随机读操作

public SelectMappedBufferResult selectMappedBuffer(int pos) {
        int readPosition = getReadPosition();
        if (pos < readPosition && pos >= 0) {
            if (this.hold()) {
                ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
                byteBuffer.position(pos);
                int size = readPosition - pos;
                ByteBuffer byteBufferNew = byteBuffer.slice();
                byteBufferNew.limit(size);
                return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this);
            }
        }

        return null;
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15

随机读的操作有两个方法，上面显示第一种方法，读取从指定位置开始的所有消息，海有一种是读取指定位置开始的指定消息大小的消息内容。这两个方法均是调用 ByteBuffer 的 slice 和 limit 方法获取消息内容，然后初始化 SelectMapedBufferResult 对象并返回；该对象的 startOffset 变量是读取消息的开始位置加上该文件的起始偏移量；

（二）MappedFileQueue 
我们在应用层中访问commitlog和consumequeue是通过MappFileQueue来操作MappedFile类的，从而间接操作磁盘上面的文件，MappFileQueue是多个MappedFile组成的队列。接下来解读一下其重要的方法 
（1）获取在指定时间点后更新的文件

public MappedFile getMappedFileByTime(final long timestamp) {
        Object[] mfs = this.copyMappedFiles(0);

        if (null == mfs)
            return null;

        for (int i = 0; i < mfs.length; i++) {
            MappedFile mappedFile = (MappedFile) mfs[i];
            if (mappedFile.getLastModifiedTimestamp() >= timestamp) {
                return mappedFile;
            }
        }

        return (MappedFile) mfs[mfs.length - 1];

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

解析：遍历MappedFile，若遇到的文件更新时间大于指定时间，则返回该对象，若找不到则返回最后一个MappedFile对象

（2）清理指定偏移量所在文件之后的文件

public void truncateDirtyFiles(long offset) {
        List<MappedFile> willRemoveFiles = new ArrayList<MappedFile>();

        for (MappedFile file : this.mappedFiles) {
            long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;
            if (fileTailOffset > offset) {
                if (offset >= file.getFileFromOffset()) {
                    file.setWrotePosition((int) (offset % this.mappedFileSize));
                    file.setCommittedPosition((int) (offset % this.mappedFileSize));
                    file.setFlushedPosition((int) (offset % this.mappedFileSize));
                } else {
                    file.destroy(1000);
                    willRemoveFiles.add(file);
                }
            }
        }

        this.deleteExpiredFile(willRemoveFiles);
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20

解析：遍历整个列表，每个MappedFile对象对应着一个固定大小的文件，当当文件的起始偏移量fileFromOffset<=offset<=fileFromOffset+fileSize，则表示指定的位置偏移量 offset 落在的该文件上，则将对应的 MapedFile 对象的 wrotepostion 和commitPosition 设置为 offset%fileSize，若文件的起始偏移量fileFromOffset>offset，即是命中的文件之后的文件，则将这些文件删除并且从 MappFileQueue 的 MapedFile 列表中清除掉。

（3）获取或创建最后一个文件

public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
        long createOffset = -1;
        MappedFile mappedFileLast = getLastMappedFile();

        //列表为空或者最后一个对象对应的文件已经写满，则创建一个新的文件（即新的 MapedFile 对象） ；
        if (mappedFileLast == null) {
            createOffset = startOffset - (startOffset % this.mappedFileSize);
        }

        if (mappedFileLast != null && mappedFileLast.isFull()) {
            createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
        }
        //若存在最后一个文件（对应最后一个 MapedFile 对象） 并且未写满，则直接返回最后一个 MapedFile 对象。
        //若列表为空，则创建的新文件的文件名（即 fileFromOffset 值）为 0；若最后一个文件写满，则新文件的文件名等于最后一个文件的fileFromOffset+fileSize
        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;
            //若在 Broker 启动初始化的时候会创建了后台服务线程（ AllocateMapedFileService 服务） ，则调用
            //AllocateMapedFileService.putRequestAndReturnMapedFile 方法，在该方法中用下一个文件的文件路径、下下一个文件的路径、文件大小为参数初始化
            //AllocateRequest 对象，并放入该服务线程的requestQueue:PriorityBlockingQueue<AllocateRequest>变量中，由该线程在
            //后台监听 requestQueue 队列，若该队列中存在 AllocateRequest 对象，则利用该对象的变量值创建 MapedFile 对象（即在磁盘中生成了对应的物理文件） ，并
            //存入 AllocateRequest 对象的 MapedFile 变量中，并且在下一个新文件之后继续将下下一个新文件也创建了
            if (this.allocateMappedFileService != null) {
                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);
                }
            }
            //最后将创建或返回的 MapedFile 对象存入 MapedFileQueue 的 MapedFile 列表中，并返回该 MapedFile 对象给调用者
            if (mappedFile != null) {
                if (this.mappedFiles.isEmpty()) {
                    mappedFile.setFirstCreateInQueue(true);
                }
                this.mappedFiles.add(mappedFile);
            }

            return mappedFile;
        }

        return mappedFileLast;
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47

（4）统计内存数据中还未刷盘的大小

public long howMuchFallBehind() {
        if (this.mappedFiles.isEmpty())
            return 0;

        long committed = this.flushedWhere;
        if (committed != 0) {
            MappedFile mappedFile = this.getLastMappedFile(0, false);
            if (mappedFile != null) {
                return (mappedFile.getFileFromOffset() + mappedFile.getWrotePosition()) - committed;
            }
        }

        return 0;
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

解析：调用 getLastMapedFile 方法获取 Mapped 队列中最后一个 MapedFile 对象，计算得出未刷盘的消息大小，计算公式为：最后一个 MapedFile 对象fileFromOffset+写入位置 wrotepostion-commitedWhere（上次刷盘的位置）

（三）Commitlog 
CommitLog是用于存储真实的物理消息的结构，ConsumeQueue是逻辑队列，仅仅存储了CommitLog的位移而已，真实的存储都在本结构中。 
commitlog文件的存储地址： $HOME\store\commitlog\${fileName}.每个文件的大小默认为1G,commitlog的文件名fileName，名字长度为20位，左边补0，剩余为起始偏移量，比如 00000000000000000000 代表了第一个文件，起始偏移量为 0， 文件大小为 1G=1073741824； 当这个文件满了，第二个文件名字为 00000000001073741824。

（1）存储结构

字段 字段大小(字节) 字段含义

msgSize	4	代表这个消息的大小
MAGICCODE	4	MAGICCODE = daa320a7
BODY CRC	4	消息体 BODY CRC
queueId	4
flag	4
QUEUEOFFSET	8	这个值是个自增值不是真正的 consume queue 的偏移量，可以代表这个consumeQueue 队列或者 tranStateTable 队列中消息的个数
SYSFLAG	4	指明消息是事物事物状态等消息特征，二进制为四个字节从右往左数：当 4 个字节均为 0（值为 0）时表示非事务消息；当第 1 个字 节为 1（值为 1）时表示表示消息是压缩的（Compressed）；当第 2 个字节为 1（值为 2） 表示多消息（MultiTags）；当第 3 个字节为 1（值为 4）时表示 prepared 消息；当第 4 个字节为 1（值为 8）时表示commit 消息； 当第 3/4 个字节均为 1 时（值为 12）时表示 rollback 消息;当第 3/4 个字节均为 0 时表 示非事务消息；
BORNTIMESTAMP	8	消息产生端(producer)的时间戳
BORNHOST	8	消息产生端(producer)地址(address:port)
STORETIMESTAMP	8	消息在broker存储时间
STOREHOSTADDRESS	8	消息存储到broker的地址(address:port)
RECONSUMETIMES	8	消息被某个订阅组重新消费了几次（订阅组 之间独立计数）,因为重试消息发送到了topic 名字为%retry%groupName的队列 queueId=0的队列中去了，成功消费一次记 录为0；
PreparedTransaction Offset	8	表示是prepared状态的事物消息
messagebodyLength	4	消息体大小值
messagebody	bodyLength	消息体内容
topicLength	1	topic名称内容大小
topic	topicLength	topic的内容值
propertiesLength	2	属性值大小
properties	propertiesLength	propertiesLength大小的属性数据

（2）正常恢复CommitLog内存数据

public void recoverNormally() {
        boolean checkCRCOnRecover = this.defaultMessageStore.getMessageStoreConfig().isCheckCRCOnRecover();
        final List<MappedFile> mappedFiles = this.mappedFileQueue.getMappedFiles();
        if (!mappedFiles.isEmpty()) {
            // 从 MapedFileQueue 的 MapedFile 列表的倒数第三个对象（即倒数第三个文件）开始遍历每块消息单元，
            //若总共没有三个文件，则从第一个文件开始遍历每块消息单元
            int index = mappedFiles.size() - 3;
            if (index < 0)
                index = 0;

            MappedFile mappedFile = mappedFiles.get(index);
            ByteBuffer byteBuffer = mappedFile.sliceByteBuffer();
            long processOffset = mappedFile.getFileFromOffset();
            long mappedFileOffset = 0;
            while (true) {
                //进行 CRC 的校验，在校验过程中，若检查到第 5 至 8 字节 MAGICCODE 字段等于 BlankMagicCode （ cbd43194）则返回 msgSize=0
                //的 DispatchRequest 对象；若校验未通过或者读取到的信息为空则返回msgSize=-1 的 DispatchRequest 对象；
                DispatchRequest dispatchRequest = this.checkMessageAndReturnSize(byteBuffer, checkCRCOnRecover);
                int size = dispatchRequest.getMsgSize();
                // 对于 msgSize 大于零，则读取的偏移量 mapedFileOffset 累加 msgSize；
                if (dispatchRequest.isSuccess() && size > 0) {
                    mappedFileOffset += size;
                }
                //若等于零，则表示读取到了文件的最后一块信息，则继续读取下一个 MapedFile
                //对象的文件； 直到消息的 CRC 校验未通过或者读取完所有信息为止
                else if (dispatchRequest.isSuccess() && size == 0) {
                    index++;
                    if (index >= mappedFiles.size()) {
                        // Current branch can not happen
                        log.info("recover last 3 physics file over, last maped file " + mappedFile.getFileName());
                        break;
                    } else {
                        mappedFile = mappedFiles.get(index);
                        byteBuffer = mappedFile.sliceByteBuffer();
                        processOffset = mappedFile.getFileFromOffset();
                        mappedFileOffset = 0;
                        log.info("recover next physics file, " + mappedFile.getFileName());
                    }
                }
                // Intermediate file read error
                else if (!dispatchRequest.isSuccess()) {
                    log.info("recover physics file end, " + mappedFile.getFileName());
                    break;
                }
            }
            //计算有效信息的最后位置 processOffset，取最后读取的MapedFile 对象的 fileFromOffset 加上最后读取的位置 mapedFileOffset 值。
            processOffset += mappedFileOffset;
            //更新数据
            this.mappedFileQueue.setFlushedWhere(processOffset);
            this.mappedFileQueue.setCommittedWhere(processOffset);
            this.mappedFileQueue.truncateDirtyFiles(processOffset);
        }
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53

（3）异常恢复CommitLog内存数据 
在 Broker 启动过程中会调用该方法。 与正常恢复的区别在于：正常恢复是从倒数第 3 个文件开始恢复；而异常恢复是从最后的文件开始往前寻找与checkpoint 文件的记录相匹配的一个文件。首先， 若该 MapedFile 队列为空，则 MapedFileQueue 对象的 commitedWhere 
等于零，并且调用 DefaultMessageStore.destroyLogics() 方法删除掉逻辑队列 consumequeue 中的物理文件以及清理内存数据。 否则从 MapedFileQueue 的MapedFile 列表找到从哪个文件（对应的 MapedFile 对象）开始恢复数据，查找 
逻辑如下： 
1、从 MapedFile 列表中的最后一个对象开始往前遍历每个 MapedFile 对象，检查该 MapedFile 对象对应的文件是否满足恢复条件，查找逻辑如下，若查找完整个队列未找到符合条件的 MapedFile 对象，则从第一个文件开始恢复。 
A） 从该文件中获取第一个消息单元的第 5 至 8 字节的 MAGICCODE 字段，若该字段等于 MessageMagicCode（即不是正常的消息内容），则直接返回后继续检查前一个文件； 
B）获取第一个消息单元的第 56 位开始的 8 个字节的 storeTimeStamp 字段，若等于零，也直接返回后继续检查前一个文件； 
C）检查是否开启消息索引功能 （ MessageStoreConfig .messageIndexEnable，默认为 true） 并且是否使用安全的消息索引功能（ MessageStoreConfig.MessageIndexSafe，默认为 false，在可靠模式下， 异常宕机恢复慢； 非可靠模式下，异常宕机恢复快） ，若开启可靠模式下面的消息索引，则消息的storeTimeStamp 字段表示的时间戳必须小于 checkpoint 文件中物理队列消息时间戳、逻辑队列消息时间戳、索引队列消息时间戳这三个时间戳中最小值，才满足恢复数据的条件； 否则消息的 storeTimeStamp 字段表示的时间戳必须小于 
checkpoint 文件中物理队列消息时间戳、逻辑队列消息时间戳这两个时间戳中最小值才满足恢复数据的条件； 
接下来跟正常恢复一致了

（4）写入消息 
Broker在收到producer的消息后，会间接调用CommitLog.putMessage(MessageExtBrokerInner msg)进行消息的写入。

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)
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        // Back to Results
        AppendMessageResult result = null;

        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

        String topic = msg.getTopic();
        int queueId = msg.getQueueId();

        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        //获取消息的 sysflag 字段，检查消息是否是非事务性
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE//
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            //获取消息延时投递时间级别
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }
                //topic值更改为 “ SCHEDULE_TOPIC_XXXX”
                topic = ScheduleMessageService.SCHEDULE_TOPIC;
                //根据延迟级别获取延时消息的队列 ID（ queueId 等于延迟级别减去 1） 并更改 queueId 值
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                //将消息中原真实的 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);
            }
        }

        long eclipseTimeInLock = 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);
            //调用 MapedFileQueue.getLastMapedFile 方法获取或者创建最后一个文件（即MapedFile 列表中的最后一个 MapedFile 对象），
            //若还没有文件或者已有的最后一个文件已经写满则创建一个新的文件，即创建一个新的 MapedFile 对象并返回
            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
                log.error("create maped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
            }
            //消息内容写入 MapedFile.mappedByteBuffer： MappedByteBuffer 对象，
            //即写入消息缓存中；由后台服务线程定时的将缓存中的消息刷盘到物理文件中
            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            //若最后一个 MapedFile 剩余空间不足够写入此次的消息内容，即返回状态为
            //END_OF_FILE 标记， 则再次调用 MapedFileQueue.getLastMapedFile 方法获取新
            //的 MapedFile 对象然后调用 MapedFile.appendMessage 方法重写写入，最后继续
            //执行后续处理操作；若为 PUT_OK 标记则继续后续处理；若为其他标记则返回错误信息给上层
            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 maped 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);
            }

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

        if (eclipseTimeInLock > 500) {
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, 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);
        handleHA(result, putMessageResult, msg);

        return putMessageResult;
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76
• 77
• 78
• 79
• 80
• 81
• 82
• 83
• 84
• 85
• 86
• 87
• 88
• 89
• 90
• 91
• 92
• 93
• 94
• 95
• 96
• 97
• 98
• 99
• 100
• 101
• 102
• 103
• 104
• 105
• 106
• 107
• 108
• 109
• 110
• 111
• 112
• 113
• 114
• 115
• 116
• 117
• 118
• 119

public void handleDiskFlush(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
        // 若该 Broker 是同步刷盘，并且消息的 property 属性中“ WAIT”参数,
        //表示是否等待服务器将消息存储完毕再返回（可能是等待刷盘完成或者等待同步复制到其他服务器） )为空或者为 TRUE
        if (FlushDiskType.SYNC_FLUSH == this.defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            final GroupCommitService service = (GroupCommitService) this.flushCommitLogService;
            if (messageExt.isWaitStoreMsgOK()) {
                //构建 GroupCommitRequest 对象，其中 nextOffset 变量的值等于wroteOffset（写入的开始物理位置）加上 wroteBytes（写入的大小） ,
                //表示下一次写入消息的开始位置
                GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
                //将该对象存入 GroupCommitService.requestsWrite 写请求队列中
                //该线程的 doCommit 方法中遍历读队列的数据，检查MapedFileQueue.committedWhere（刷盘刷到哪里的记录）
                service.putRequest(request);
                boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
                if (!flushOK) {
                    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();
            }
        }
        //若该 Broker 为异步刷盘（ASYNC_FLUSH） ，唤醒 FlushRealTimeService 线程服务
        else {
            if (!this.defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                flushCommitLogService.wakeup();
            } else {
                commitLogService.wakeup();
            }
        }
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31

 public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
        //若该Broker为同步双写主用（ SYNC_MASTER）
        if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
            HAService service = this.defaultMessageStore.getHaService();
            //并且消息的 property 属性中“ WAIT”参数为空或者为 TRUE，则等待监听主 Broker 将数据同步到从 Broker
            //的结果，若同步失败，则置 PutMessageResult 对象的 putMessageStatus 变量为FLUSH_SLAVE_TIMEOUT
            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();
                    boolean flushOK =
                        request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());
                    if (!flushOK) {
                        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
                else {
                    // Tell the producer, slave not available
                    putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE);
                }
            }
        }

    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29

（5）DefaultAppendMessageCallback 
该类实现了 AppendMessageCallback 回调类的 doAppend，顺序写方法的具体逻辑如下： 
1）获取当前内存对象的写入位置（ wrotePostion 变量值）；若写入位置没有超过文件大小则继续顺序写入； 
2）由内存对象 mappedByteBuffer 创建一个指向同一块内存的 ByteBuffer对象，并将内存对象的写入指针指向写入位置； 
3）以文件的起始偏移量（ fileFromOffset）、 ByteBuffer 对象、该内存对象剩余的空间（ fileSize-wrotePostion）、消息对象 msg 为参数调用

（四）Consumequeue 
对应每个topic和queueid下面的所有文件，默认存储位置为$HOME/store/consumequeue/{topic}/{queueId}/{fileName}，每条数据的结构如下： 
消息起始物理偏移量(physical offset, long 8字节)+消息大小(int,4字节)+tagsCode(long 8字节) 
每个 cosumequeue 文件的名称 fileName，名字长度为 20 位，左边补零，剩余为起始偏量； 比如 00000000000000000000 代表了第一个文件，起始偏移量为 0，文件大小为 600W， 当第一个文件满之后创建的第二个文件的名字为00000000000006000000，起始偏移量为6000000，以此类推，第三个文件名字为00000000000012000000，起始偏移量12000000。消息存储的时候会顺序写入文件，当文件满了，写入下一个文件。

该类具有下面几个主要的方法 
1，删除指定偏移量之后的逻辑文件（ truncateDirtyLogicFiles） 
2，恢复 ConsumeQueue 内存数据（ recover） 
3，查找消息发送时间最接近 timestamp 逻辑队列的 offset 
4，获取最后一条消息对应物理队列的下一个偏移量（ getLastOffset） 
5，消息刷盘（ commit） 
6，将 commitlog 物理偏移量、消息大小等信息直接写入 consumequeue 中 
7，根据消息序号索引获取 consumequeue 数据（ getIndexBuffer） 
8，根据物理队列最小 offset计算修正逻辑队列最小 offset （ correctMinOffset） 
9，获取指定位置所在文件的下一个文件的起始偏移量（ rollNextFile）

（六）IndexFile 
为操作Index文件提供访问服务，存储位置：$HOME \store\index\${fileName}，文件名 
fileName 是以创建时的时间戳命名的，文件大小是固定的。 

各个字段解释 
Index Header 结构各字段的含义： 
beginTimestamp：第一个索引消息落在 Broker 的时间戳； 
endTimestamp：最后一个索引消息落在 Broker 的时间戳； 
beginPhyOffset：第一个索引消息在 commitlog 的偏移量； 
endPhyOffset：最后一个索引消息在 commitlog 的偏移量； 
hashSlotCount：构建索引占用的槽位数； 
indexCount：构建的索引个数；

Slot Table 里面的每一项保存的是这个 topic-key 是第几个索引；根据 
topic-key 的 Hash 值除以 500W 取余得到这个 Slot Table 的序列号，然后将此 
索引的顺序个数存入此 Table 中。 
Slottable 的位置（ absSlotPos）的计算公式： 40+keyHash%（ 500W） *4；

Index Linked List 的字段含义： 
keyHash:topic-key(key 是消息的 key)的 Hash 值； 
phyOffset:commitLog 真实的物理位移； 
timeOffset：时间位移，消息的存储时间与 Index Header 中 beginTimestamp 
的时间差； 
slotValue：当 topic-key(key 是消息的 key)的 Hash 值取 500W 的余之后得 
到的 Slot Table 的 slot 位置中已经有值了（即 Hash 值取余后在 Slot Table 
中有冲突时），则会用最新的 Index 值覆盖，并且将上一个值写入最新 Index 
的 slotValue 中，从而形成了一个链表的结构。 
Index Linked List 的位置（ absIndexPos）的计算公式： 40+ 500W*4+index 
的顺序数*40；

（1）向index文件写入索引消息

public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        if (this.indexHeader.getIndexCount() < this.indexNum) {
            int keyHash = indexKeyHashMethod(key);
            int slotPos = keyHash % this.hashSlotNum;
            //首先根据 key 的 Hash 值计算出 absSlotPos 值；
            int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;

            FileLock fileLock = null;

            try {

                // fileLock = this.fileChannel.lock(absSlotPos, hashSlotSize,
                // false);
                //根据 absSlotPos值作为 index文件的读取开始偏移量读取 4个字节的值，
                //即为了避免 KEY 值的 hash 冲突，将之前的 key 值的索引顺序数给冲突了，故先
                //从 slot Table 中的取当前存储的索引顺序数，若该值小于零或者大于当前的索
                //引总数（ IndexHeader 的 indexCount 值）则视为无效，即置为 0；否则取出该位
                //置的值，放入当前写入索引消息的 Index Linked 的 slotValue 字段中；
                int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {
                    slotValue = invalidIndex;
                }
                //计算当前存时间距离第一个索引消息落在 Broker 的时间戳beginTimestamp 的差值，放入当前写入索引消息的 Index Linked 的 timeOffset字段中
                long timeDiff = storeTimestamp - this.indexHeader.getBeginTimestamp();

                timeDiff = timeDiff / 1000;

                if (this.indexHeader.getBeginTimestamp() <= 0) {
                    timeDiff = 0;
                } else if (timeDiff > Integer.MAX_VALUE) {
                    timeDiff = Integer.MAX_VALUE;
                } else if (timeDiff < 0) {
                    timeDiff = 0;
                }
                //计算 absIndexPos 值，然后根据数据结构上值写入 Index Linked 中；
                int absIndexPos =
                    IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                        + this.indexHeader.getIndexCount() * indexSize;

                this.mappedByteBuffer.putInt(absIndexPos, keyHash);
                this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
                this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
                //将索引总数写入 slot Table 的 absSlotPos 位置；
                this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
                //若为第一个索引，则更新 IndexHeader 的 beginTimestamp 和beginPhyOffset 字段；
                if (this.indexHeader.getIndexCount() <= 1) {
                    this.indexHeader.setBeginPhyOffset(phyOffset);
                    this.indexHeader.setBeginTimestamp(storeTimestamp);
                }
                //更新 IndexHeader 的 endTimestamp 和 endPhyOffset 字段；
                //将 IndexHeader 的 hashSlotCount 和 indexCount 字段值加 1；
                this.indexHeader.incHashSlotCount();
                this.indexHeader.incIndexCount();
                this.indexHeader.setEndPhyOffset(phyOffset);
                this.indexHeader.setEndTimestamp(storeTimestamp);

                return true;
            } catch (Exception e) {
                log.error("putKey exception, Key: " + key + " KeyHashCode: " + key.hashCode(), e);
            } finally {
                if (fileLock != null) {
                    try {
                        fileLock.release();
                    } catch (IOException e) {
                        log.error("Failed to release the lock", e);
                    }
                }
            }
        } else {
            log.warn("Over index file capacity: index count = " + this.indexHeader.getIndexCount()
                + "; index max num = " + this.indexNum);
        }

        return false;
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68
• 69
• 70
• 71
• 72
• 73
• 74
• 75
• 76

（七）IndexService 
IndexService 是线程类服务，在启动 Broker 时启动该线程服务。 该类主要有两个功能，第一，是定时的创建消息的索引；第二是为应用层提供访问 index索引文件的接口

（1）（重点）创建消息的索引

public void buildIndex(DispatchRequest req) {
        //获取index文件,逻辑如下
        /*  A）从 IndexFile 列表中获取最后一个 IndexFile 对象；若该对象对应的
            Index 文件没有写满，即 IndexHeader 的 indexCount 不大于 2000W；则直接返回
            该对象；
            B）若获得的该对象为空或者已经写满，则创建新的 IndexFile 对象，即
            新的 Index 文件，若是因为写满了而创建，则在创建新 Index 文件时将该写满的
            Index 文件的 endPhyOffset 和 endTimestamp 值初始化给新 Index 文件中
            IndexHeader 的 beginPhyOffset 和 beginTimestamp。
            C）启一个线程，调用 IndexFile 对象的 fush 将上一个写满的 Index 文
            件持久化到磁盘物理文件中；然后更新 StoreCheckpoint.IndexMsgTimestamp
            为该写满的 Index 文件中 IndexHeader 的 endTimestamp；
        */
        IndexFile indexFile = retryGetAndCreateIndexFile();
        if (indexFile != null) {
            /*  遍历 requestQueue 队列中的请求消息。 将每个请求消息的
                commitlogOffset 值与获取的 IndexFile 文件的 endPhyOffset 进行比较，若小
                于 endPhyOffset 值，则直接忽略该条请求信息； 对于消息类型为 Prepared 和
                RollBack 的也直接忽略掉
            */
            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_PREPARED_TYPE:
                case MessageSysFlag.TRANSACTION_COMMIT_TYPE:
                    break;
                case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE:
                    return;
            }
            /*
                对于一个 topic 可以有多个 key 值，每个 key 值以空格分隔，遍历每个
                key 值，将 topic-key 值作为 putKey 方法的入参 key 值，将该 topic 的物理偏
                移量存入 Index 文件中，若存入失败则再次获取 IndexFile 对象重复调用 putKey
                方法。
            */
            if (req.getUniqKey() != null) {
                indexFile = putKey(indexFile, msg, buildKey(topic, req.getUniqKey()));
                if (indexFile == null) {
                    log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                    return;
                }
            }

            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(indexFile, msg, buildKey(topic, key));
                        if (indexFile == null) {
                            log.error("putKey error commitlog {} uniqkey {}", req.getCommitLogOffset(), req.getUniqKey());
                            return;
                        }
                    }
                }
            }
        } else {
            log.error("build index error, stop building index");
        }
    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54
• 55
• 56
• 57
• 58
• 59
• 60
• 61
• 62
• 63
• 64
• 65
• 66
• 67
• 68

（八）Config 
在$HOME\store\config 目录下面存储各类 config 文件，包括：consumerOffset.json、 delayOffset.json、 subscriptionGroup.json、topics.json 四类 config 文件。topics.json 文件由 TopicConfigManager 类解析并存储； 存储每个 topic的读写队列数、权限、是否顺序等信息。consumerOffset.json 文件由 ConsumerOffsetManager 类解析并存储； 存储每个消费者 Consumer 在每个 topic 上对于该 topic 的 consumequeue 队列的消费进度；delayOffset.json 文件由 ScheduleMessageService 类解析并存储； 存储对于延迟主题SCHEDULE_TOPIC_XXXX 的每个 consumequeue 队列的消费进度；subscriptionGroup.json 文件由 SubscriptionGroupManager 类解析并存储；存储每个消费者 Consumer 的订阅信息。

（九）DefaulMessageStore 
这是所有文件的访问入口，代码过于长，这里就跳几个重要的方法讲解 
（1） findConsumeQueue。 
根据 topic 和 queueId 查找 ConsumeQueue。 
1）先根据 topic 查找 values 值,若没有则创建此类型的 Map 并存入 ConsumeQueue 集合； 
2）从获得的 Map 对象中根据 queueId 获得 ConsumeQueue 对象，若为空，则创建 ConsumeQueue 对象，在创建该对象时初始化了 topic、queueId、storePath、mapedFileSize 等变量；也初始化了 MapedFileQueue 对象； 将初始化的ConsumeQueue 对象存入 Map 中并返回；若不为空则直接返回。

（2）将消息写入commitlog中 
1）检查是否 shutdown，若是则直接返回服务不可用的错误； 
2）检查是否为备用 Broker，若是则直接返回服务不可用的错误； 
3）检查是否有写的权限，若是则直接返回服务不可用的错误； 
4）检查 topic 的长度是否大于最大值 127，若是则返回消息不合法； 
5）检查 peroperties 的长度是否大于 32767，若是则返回消息不合法； 
6）上述检查全部通过之后，调用 CommitLog 对象的 putMessage 方法进行消 
息的写入； 
7）完成消息写入之后，调用 StoreStatsService 对象进行相关统计工作；

（3）读取commitlog消息 
大致逻辑如下： 
1、 调用 CommitLog.getMaxOffset()方法获取 commitlog 文件目前写入的最大位置 maxOffsetPy； 
2、 以 topic 和 queueId 为参数调用 findConsumeQueue 方法， 从DefaultMessageStore.consumeQueueTable 中获取 ConsumeQueue 对象 
3、 获取该 ConsumeQueue 对象的最小逻辑 offset（命名：minOffset=minLogicOffset/20）和最大逻辑 offset（命名：maxOffset=MapedFileQueue.getMaxOffset()/20）； 
4、对请求参数中的 queueoffset 进行有效性检验：若请求消息中的 queueoffset 在 minOffset 与 maxOffset 之间，则继续执行后续操作步骤读取消息 
5、根据 queueoffset 为开始读取位置调用 getIndexBuffer(final longstartIndex)方法获取 consumequeue 数据；若获取到了数据，则继续下面的执行步骤； 
6、 遍历从 ConsumeQueue 获取的数据中的每个数据单元在遍历开始时设计一个标志位 nextPhyFileStartOffset，初始为 Long.MIN_VALUE，用于检测每个数据单元所保存的物理偏移量的 commitlog 数据是否还存在，若不存在则继续解析下一个数据单元 
7、 计算下一次读取数据时的开始偏移量 NextBeginOffset，计算方式是：请求消息中的 offset+上面遍历完之后最后一个 offset 值除以 20； 
8、检查未拉取的消息的大小是否大于最大可使用内存，若大于，则建议从备用 Broker 拉取消息，即设置 GetMessageResult.suggestPullingFromSlave等于 true；未拉取的消息的大小计算方式是： commitlog 的最大物理偏移 offset减去此次拉取的最后一个消息的物理偏移 offset 即为还未拉取的消息大小 
9、若 status=FOUND，则将 StoreStatsService.getMessageTimesTotalFound变量加 1；否则将 StoreStatsService.getMessageTimesTotalMiss 变量加 1； 
10、返回 GetMessageResult 对象，该对象中包括 status、NextBeginOffset、consumequeue 中的最大 offset 和最小 offset 等信息；

（4）DefaultMessageStore.ReputMessageService 服务线程 
该服务线程会一直不间断的监听 reputFromOffset 偏移量之后的 commitlog数据， 若 commitlog 数据有增加， 即 reputFromOffset 偏移量之后新增了数据，则获取出来并创建 consumequeue 和 index

（5）DefaultMessageStore.DispatchMessageService 服务线程 
该线程会负责处理 DispatchRequest 请求，为请求中的信息创建consumequeue 数据和 index 索引。 当调用者调用DispatchMessageService.putRequest(DispatchRequest dispatchRequest)方 
法时，即唤醒该线程，并将 requestsWrite 队列和 requestsRead 队列互换，其中 requestsRead 队列一般是空的，然后遍历 requestsRead 队列

（十）Abort 
启动 Broker 时，在目录$HOME \store 下面创建 abort 文件，没有任何内容；只是标记是否正常关闭，若为正常关闭，则在关闭时会删掉此文件；若未正常关闭则此文件一直保留，下次启动时根据是否存在此文件进行不同方式的内存数据恢复

（十一）Checkpoint 
Checkpoint 文件的存储路径默认为$HOME/store/checkpoint，文件名就是checkpoint，文件的数据结构如下： 
physicMsgTimestamp(8)+logicsMsgTimestamp(8)+ indexMsgTimestamp(8)

Checkpoint 文件由 StoreCheckpoint 类来解析并存储内容。在进行消息写入commitlog，物理偏移量/消息大小写入 consumequeue、创建 Index 索引这三个操作之后都要分别更新 physicMsgTimestamp、 logicsMsgTimestamp、indexMsgTimestamp 字段的内容； 
在从异常中恢复 commitlog 内存数据时要用得该文件的三个字段的时间戳。

（十二）HA高可用 
在集群模式下，为了保证高可用，必须要保证备用 Broker 与主用 Broker 信息是一致的， 在备用 Broker 初始化时设置的了定时任务，每个 60 秒调用SlaveSynchronize.syncAll()方法发起向主用 Broker进行一次 config类文件的同步，而消息数据的同步由主备 Broker 通过心跳检测的方式完成，每隔 5 秒进行一次心跳。 主用 Broker 提供读写服务，而备用 Broker 只提供读服务。 
同步的信息包括了：Topic配置，消费进度信息，延迟消费进度信息，订阅关系，消息数据 
（1）主备Broker异步同步流程 

（2）主备Broker同步双写流程图 

（十三）事务相关的文件 
有 TransactionStateService 类，用于存储每条事务消息的状态。 在该类中有两个成员变量 tranRedoLog:ConsumeQueue 和tranStateTable: MapedFileQueue,其中 tranRedoLog 变量用于事务状态的Redolog，当进程意外宕掉，可通过 redolog 恢复所有事务的状态，tranStateTable 变量用于存储事务消息的事务状态。

（1）事务消息状态文件 
tranStateTable 变量对应的物理文件的存储路径默认为$HOME/store/transaction/statetable/{fileName}， 每个文件由 2000000 条数据组成， 每条 
数据总共有 24 个字节，其结构如下，因此每个文件的大小为 2000000*24； 
 
每个 statetable 文件的名称 fileName，名字长度为 20 位，左边补零，剩余为起始偏移量； 比如 00000000000000000000 代表了第一个文件，起始偏移量为0，文件大小为2000000*24， 当第一个文件满之后创建的第二个文件的名字为00000000000048000000，起始偏移量为 48000000，以此类推，第三个文件名字为 00000000000096000000，起始偏移量为 96000000， 消息存储的时候会顺序写入文件，当文件满了，写入下一个文件。

（2）事务消息 REDO 日志文件 
tranRedoLog 变量对应的物理文件的存储路径默认为$HOME/store/transaction/redolog/TRANSACTION_REDOLOG_TOPIC_XXXX/0/{fileName}， 每个文件由 2000000 条数据组成， 每条数据总共有 20 个字节，因此每个文件的大小 
为 2000000*20，每条数据的结构与 ConsumeQueue 文件一样，相当于 topic 值等于“ TRANSACTION_REDOLOG_TOPIC_XXXX”， queueId 等于 0

（3）定期向 Producer 回查事务消息的最新状态 
在 Broker 启动 TransactionStateService 时，为每个 tranStateTable 文件创建了一个定时任务，该定时任务是每隔 1 分钟遍历一遍文件中的所有消息，对于处于 PREPARED 状态的事务消息，向 Producer 发送回查请求获取最新的事务状态

 

 

 

http://blog.csdn.net/KilluaZoldyck/article/details/76775397