# Nicksxs's Blog

What hurts more, the pain of hard work or the pain of regret?

0%

# 聊一下 RocketMQ 的消息存储二

Posted on 2021-09-12 In MQ , RocketMQ , 消息队列 Views: Views: Disqus:

### CommitLog 结构

CommitLog 是 rocketmq 的服务端，也就是 broker 存储消息的的文件，跟 kafka 一样，也是顺序写入，当然消息是变长的，生成的规则是每个文件的默认1G =1024 * 1024 * 1024，commitlog的文件名fileName，名字长度为20位，左边补零，剩余为起始偏移量；比如00000000000000000000代表了第一个文件，起始偏移量为0，文件大小为1G=1 073 741 824Byte；当这个文件满了，第二个文件名字为00000000001073741824，起始偏移量为1073741824, 消息存储的时候会顺序写入文件，当文件满了则写入下一个文件，代码中的定义

    // CommitLog file size,default is 1G
    private int mapedFileSizeCommitLog = 1024 * 1024 * 1024;

本地跑个 demo 验证下，也是这样，这里奇妙有几个比较巧妙的点（个人观点），首先文件就刚好是 1G，并且按照大小偏移量去生成下一个文件，这样获取消息的时候按大小算一下就知道在哪个文件里了，

代码中写入 CommitLog 的逻辑可以从这开始看

    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());
            if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
                || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
                // Delay Delivery
                if (msg.getDelayTimeLevel() > 0) {
                    if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                        msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                    }

                    topic = ScheduleMessageService.SCHEDULE_TOPIC;
                    queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                    // Backup real 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);

                if (null == mappedFile || mappedFile.isFull()) {
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
                }
                if (null == mappedFile) {
                    log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
                }

                result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                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 mapped 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;
        }

前面也看到在CommitLog 目录下是有大小为 1G 的文件组成，在实现逻辑中，其实是通过 `org.apache.rocketmq.store.MappedFileQueue` ，内部是存的一个`MappedFile`的队列，对于写入的场景每次都是通过`org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile()` 获取最后一个文件，如果还没有创建，或者最后这个文件已经满了，那就调用 `org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile(long)`

    public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
            long createOffset = -1;
      			// 调用前面的方法，只是从 mappedFileQueue 获取最后一个
            MappedFile mappedFileLast = getLastMappedFile();

            // 如果为空，计算下创建的偏移量
            if (mappedFileLast == null) {
                createOffset = startOffset - (startOffset % this.mappedFileSize);
            }

    				// 如果不为空，但是当前的文件写满了
            if (mappedFileLast != null && mappedFileLast.isFull()) {
                // 前一个的偏移量加上单个文件的偏移量，也就是 1G
                createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
            }

            if (createOffset != -1 && needCreate) {
                // 根据 createOffset 转换成文件名进行创建
                String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
                String nextNextFilePath = this.storePath + File.separator
                    + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
                MappedFile mappedFile = null;

              	// 这里如果allocateMappedFileService 存在，就提交请求
                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);
                    }
                }

                if (mappedFile != null) {
                    if (this.mappedFiles.isEmpty()) {
                        mappedFile.setFirstCreateInQueue(true);
                    }
                    this.mappedFiles.add(mappedFile);
                }

                return mappedFile;
            }

            return mappedFileLast;
        }

首先看下直接创建的，

    public MappedFile(final String fileName, final int fileSize) throws IOException {
            init(fileName, fileSize);
        }
    private void init(final String fileName, final int fileSize) throws IOException {
            this.fileName = fileName;
            this.fileSize = fileSize;
            this.file = new File(fileName);
            this.fileFromOffset = Long.parseLong(this.file.getName());
            boolean ok = false;

            ensureDirOK(this.file.getParent());

            try {
              // 通过 RandomAccessFile 创建 fileChannel
                this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
              // 做 mmap 映射
                this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
                TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
                TOTAL_MAPPED_FILES.incrementAndGet();
                ok = true;
            } catch (FileNotFoundException e) {
                log.error("create file channel " + this.fileName + " Failed. ", e);
                throw e;
            } catch (IOException e) {
                log.error("map file " + this.fileName + " Failed. ", e);
                throw e;
            } finally {
                if (!ok && this.fileChannel != null) {
                    this.fileChannel.close();
                }
            }
        }

如果是提交给`AllocateMappedFileService`的话就用到了一些异步操作

    public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
            int canSubmitRequests = 2;
            if (this.messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool()
                    && BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) { //if broker is slave, don't fast fail even no buffer in pool
                    canSubmitRequests = this.messageStore.getTransientStorePool().remainBufferNumbs() - this.requestQueue.size();
                }
            }
    				// 将请求放在 requestTable 中
            AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
            boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
            // requestTable 使用了 concurrentHashMap，用文件名作为 key，防止并发
            if (nextPutOK) {
                // 这里判断了是否可以提交到 TransientStorePool，涉及读写分离，后面再细聊
                if (canSubmitRequests <= 0) {
                    log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
                        "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                    this.requestTable.remove(nextFilePath);
                    return null;
                }
              // 塞到阻塞队列中
                boolean offerOK = this.requestQueue.offer(nextReq);
                if (!offerOK) {
                    log.warn("never expected here, add a request to preallocate queue failed");
                }
                canSubmitRequests--;
            }

            // 这里的两个提交我猜测是为了多生成一个 CommitLog，
            AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
            boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
            if (nextNextPutOK) {
                if (canSubmitRequests <= 0) {
                    log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
                        "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                    this.requestTable.remove(nextNextFilePath);
                } else {
                    boolean offerOK = this.requestQueue.offer(nextNextReq);
                    if (!offerOK) {
                        log.warn("never expected here, add a request to preallocate queue failed");
                    }
                }
            }

            if (hasException) {
                log.warn(this.getServiceName() + " service has exception. so return null");
                return null;
            }

            AllocateRequest result = this.requestTable.get(nextFilePath);
            try {
              // 这里就异步等着
                if (result != null) {
                    boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
                    if (!waitOK) {
                        log.warn("create mmap timeout " + result.getFilePath() + " " + result.getFileSize());
                        return null;
                    } else {
                        this.requestTable.remove(nextFilePath);
                        return result.getMappedFile();
                    }
                } else {
                    log.error("find preallocate mmap failed, this never happen");
                }
            } catch (InterruptedException e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
            }

            return null;
        }

而真正去执行文件操作的就是 `AllocateMappedFileService`的 run 方法

    public void run() {
            log.info(this.getServiceName() + " service started");

            while (!this.isStopped() && this.mmapOperation()) {

            }
            log.info(this.getServiceName() + " service end");
        }
    private boolean mmapOperation() {
            boolean isSuccess = false;
            AllocateRequest req = null;
            try {
              // 从阻塞队列里获取请求
                req = this.requestQueue.take();
                AllocateRequest expectedRequest = this.requestTable.get(req.getFilePath());
                if (null == expectedRequest) {
                    log.warn("this mmap request expired, maybe cause timeout " + req.getFilePath() + " "
                        + req.getFileSize());
                    return true;
                }
                if (expectedRequest != req) {
                    log.warn("never expected here,  maybe cause timeout " + req.getFilePath() + " "
                        + req.getFileSize() + ", req:" + req + ", expectedRequest:" + expectedRequest);
                    return true;
                }

                if (req.getMappedFile() == null) {
                    long beginTime = System.currentTimeMillis();

                    MappedFile mappedFile;
                    if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                        try {
                          // 通过 transientStorePool 创建
                            mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
                            mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                        } catch (RuntimeException e) {
                            log.warn("Use default implementation.");
                          // 默认创建
                            mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                        }
                    } else {
                      // 默认创建
                        mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
                    }

                    long eclipseTime = UtilAll.computeEclipseTimeMilliseconds(beginTime);
                    if (eclipseTime > 10) {
                        int queueSize = this.requestQueue.size();
                        log.warn("create mappedFile spent time(ms) " + eclipseTime + " queue size " + queueSize
                            + " " + req.getFilePath() + " " + req.getFileSize());
                    }

                    // pre write mappedFile
                    if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()
                        .getMapedFileSizeCommitLog()
                        &&
                        this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
                        mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
                            this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
                    }

                    req.setMappedFile(mappedFile);
                    this.hasException = false;
                    isSuccess = true;
                }
            } catch (InterruptedException e) {
                log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
                this.hasException = true;
                return false;
            } catch (IOException e) {
                log.warn(this.getServiceName() + " service has exception. ", e);
                this.hasException = true;
                if (null != req) {
                    requestQueue.offer(req);
                    try {
                        Thread.sleep(1);
                    } catch (InterruptedException ignored) {
                    }
                }
            } finally {
                if (req != null && isSuccess)
                  // 通知前面等待的
                    req.getCountDownLatch().countDown();
            }
            return true;
        }

Related Posts

- 介绍一下 RocketMQ
- 聊一下 RocketMQ 的消息存储之 MMAP
- 聊一下 RocketMQ 的顺序消息
- 聊一下 RocketMQ 的 NameServer 源码
- 聊聊 RocketMQ 的 Broker 源码