RocketMq4.7源码解析之六(映射文件与存储文件)

旧城等待， 2022-12-02 15:19 223阅读 0赞

# 映射文件 #

RocketMQ 使用MappedFile 、MappedFileQueue 来封装存储文件  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]

## MappedFileQueue 映射文件队列 ##

字段属性

//存储目录
        private final String storePath;
        //单个文件的存储大小。
        private final int mappedFileSize;
        //mappedFiles文件集合。
        private final CopyOnWriteArrayList<MappedFile> mappedFiles = new CopyOnWriteArrayList<MappedFile>();
        //创建MmappedFiles服务类。
        private final AllocateMappedFileService allocateMappedFileService;
        //当前刷盘指针， 表示该指针之前的所有数据全部持久化到磁盘
        private long flushedWhere = 0;
        //当前数据提交指针，内存中ByteBuffer 当前的写指针，该值大于等于flushedWhere 。
        private long committedWhere = 0;

常用方法

//根据消息存储时间戳来查找MappdFile
        public MappedFile getMappedFileByTime(final long timestamp) {
        
            Object[] mfs = this.copyMappedFiles(0);
    
            if (null == mfs)
                return null;
            //遍历MappedFile
            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];
        }
    
      //删除offset 之后的所有文件
        public void truncateDirtyFiles(long offset) {
        
            List<MappedFile> willRemoveFiles = new ArrayList<MappedFile>();
    
            //遍历目录下文件
            for (MappedFile file : this.mappedFiles) {
        
                //获取文件尾部偏移
                long fileTailOffset = file.getFileFromOffset() + this.mappedFileSize;
                //若文件尾部偏移>offset
                if (fileTailOffset > offset) {
        
                    //文件开始偏移>=offset
                    //说明当前文件包含了有效偏移
                    if (offset >= file.getFileFromOffset()) {
        
                        file.setWrotePosition((int) (offset % this.mappedFileSize));
                        file.setCommittedPosition((int) (offset % this.mappedFileSize));
                        file.setFlushedPosition((int) (offset % this.mappedFileSize));
                    } else {
        
                        //说明该文件是有效文件后面创建的
                        //释放MappedFile 占用的内存资源（内存映射与内存通道等）
                        file.destroy(1000);
                        //加入待删除集合
                        willRemoveFiles.add(file);
                    }
                }
            }
    
            //删除文件
            this.deleteExpiredFile(willRemoveFiles);
        }
    
    
      public boolean load() {
        
            //加载$ ｛ ROCKET_HOME }/store/commitlog 目录下所有文件
            File dir = new File(this.storePath);
            File[] files = dir.listFiles();
            if (files != null) {
        
                // ascending order
                //按照文件名排序
                Arrays.sort(files);
                for (File file : files) {
        
    
                    //如果文件大小与配置文件的单个文件大小不一致，将忽略该目录下所有文件，
                    if (file.length() != this.mappedFileSize) {
        
                        log.warn(file + "\t" + file.length()
                            + " length not matched message store config value, please check it manually");
                        return false;
                    }
    
                    try {
        
                        //创建MappedFile 对象
                        MappedFile mappedFile = new MappedFile(file.getPath(), mappedFileSize);
    
                        //将wrotePosition 、flushedPosition ,
                        //committedPosition 三个指针都设置为文件大小。
                        mappedFile.setWrotePosition(this.mappedFileSize);
                        mappedFile.setFlushedPosition(this.mappedFileSize);
                        mappedFile.setCommittedPosition(this.mappedFileSize);
                        this.mappedFiles.add(mappedFile);
                        log.info("load " + file.getPath() + " OK");
                    } catch (IOException e) {
        
                        log.error("load file " + file + " error", e);
                        return false;
                    }
                }
            }
    
            return true;
        }
    
        /**
         *  查找最后一个文件
         * @param startOffset
         * @param needCreate 如果找不到是否需要创建新文件
         * @return
         */
        public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
        
            long createOffset = -1;
            MappedFile mappedFileLast = getLastMappedFile();
    
            //创建的偏移必须是mappedFileSize设置的倍数
            if (mappedFileLast == null) {
        
                createOffset = startOffset - (startOffset % this.mappedFileSize);
            }
    
            if (mappedFileLast != null && mappedFileLast.isFull()) {
        
                createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
            }
    
            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) {
        
                    //分别异步创建2个文件
                    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) {
        
                    //若mappedFiles队列为空
                    if (this.mappedFiles.isEmpty()) {
        
                        //设置是MappedFileQueue 队列中第一个文件
                        mappedFile.setFirstCreateInQueue(true);
                    }
                    //添加mapppedFiles集合
                    this.mappedFiles.add(mappedFile);
                }
    
                return mappedFile;
            }
    
            return mappedFileLast;
        }
    
       /**
         *  获取最后一个文件,找不到则创建
         */
        public MappedFile getLastMappedFile(final long startOffset) {
        
            return getLastMappedFile(startOffset, true);
        }
    
        /**
         * 获取最后一个文件
         */
        public MappedFile getLastMappedFile() {
        
            MappedFile mappedFileLast = null;
    
            while (!this.mappedFiles.isEmpty()) {
        
                try {
        
                    ///获取尾端文件
                    mappedFileLast = this.mappedFiles.get(this.mappedFiles.size() - 1);
                    break;
                } catch (IndexOutOfBoundsException e) {
        
                    //continue;
                } catch (Exception e) {
        
                    log.error("getLastMappedFile has exception.", e);
                    break;
                }
            }
    
            return mappedFileLast;
        }
    
    
      //获取存储文件最小偏移量
        public long getMinOffset() {
        
    
            if (!this.mappedFiles.isEmpty()) {
        
                try {
        
                    //返回第一个映射文件的起始偏移.
                    return this.mappedFiles.get(0).getFileFromOffset();
                } catch (IndexOutOfBoundsException e) {
        
                    //continue;
                } catch (Exception e) {
        
                    log.error("getMinOffset has exception.", e);
                }
            }
            return -1;
        }
    
        //获取存储文件的最大偏移量
        public long getMaxOffset() {
        
            MappedFile mappedFile = getLastMappedFile();
            if (mappedFile != null) {
        
                //返回当前文件起始偏移+可读偏移
                return mappedFile.getFileFromOffset() + mappedFile.getReadPosition();
            }
            return 0;
        }
    
        //返回存储文件当前的写指针。
        public long getMaxWrotePosition() {
        
            MappedFile mappedFile = getLastMappedFile();
            if (mappedFile != null) {
        
                //返回最后一个文件的fil eF rom Offset 加上当前写指针位置。
                return mappedFile.getFileFromOffset() + mappedFile.getWrotePosition();
            }
            return 0;
        }
    
    
     /**
         * 执行文件销毁与删除
         */
        public int deleteExpiredFileByTime(final long expiredTime,
            final int deleteFilesInterval,
            final long intervalForcibly,
            final boolean cleanImmediately) {
        
            Object[] mfs = this.copyMappedFiles(0);
    
            if (null == mfs)
                return 0;
    
            int mfsLength = mfs.length - 1;
            int deleteCount = 0;
            List<MappedFile> files = new ArrayList<MappedFile>();
            if (null != mfs) {
        
                //从倒数第二个文件开始遍历
                for (int i = 0; i < mfsLength; i++) {
        
                    MappedFile mappedFile = (MappedFile) mfs[i];
                    //计算文件的最大存活时间（ ＝ 文件的最后一次更新时间＋文件存活时间（默认72 小时）） ，
                    long liveMaxTimestamp = mappedFile.getLastModifiedTimestamp() + expiredTime;
                    if (//当前时间大于文件的最大存活
                            System.currentTimeMillis() >= liveMaxTimestamp ||
                         //需要强制删除文件(当磁盘使用超过设定的阔值)
                            cleanImmediately) {
        
                        //清除MappedFile 占有的相关资源
                        if (mappedFile.destroy(intervalForcibly)) {
        
                            //若执行成,将该文件加入到待删除文件列表中
                            files.add(mappedFile);
                            deleteCount++;
    
                            if (files.size() >= DELETE_FILES_BATCH_MAX) {
        
                                break;
                            }
    
                            if (deleteFilesInterval > 0 && (i + 1) < mfsLength) {
        
                                try {
        
                                    Thread.sleep(deleteFilesInterval);
                                } catch (InterruptedException e) {
        
                                }
                            }
                        } else {
        
                            break;
                        }
                    } else {
        
                        //avoid deleting files in the middle
                        break;
                    }
                }
            }
    
            //将文件从物理磁盘中删除。
            deleteExpiredFile(files);
    
            return deleteCount;
        }
    
     /**
         * Finds a mapped file by offset.
         * 根据消息偏移量offset 查找MappedFile
         * @param offset Offset.
         * @param returnFirstOnNotFound If the mapped file is not found, then return the first one.
         * @return Mapped file or null (when not found and returnFirstOnNotFound is <code>false</code>).
         */
        public MappedFile findMappedFileByOffset(final long offset, final boolean returnFirstOnNotFound) {
        
            try {
        
                MappedFile firstMappedFile = this.getFirstMappedFile();
                MappedFile lastMappedFile = this.getLastMappedFile();
                if (firstMappedFile != null && lastMappedFile != null) {
        
                    //偏移不在起始和结束文件中,说明越界.
                    if (offset < firstMappedFile.getFileFromOffset() || offset >= lastMappedFile.getFileFromOffset() + this.mappedFileSize) {
        
                        LOG_ERROR.warn("Offset not matched. Request offset: {}, firstOffset: {}, lastOffset: {}, mappedFileSize: {}, mappedFiles count: {}",
                            offset,
                            firstMappedFile.getFileFromOffset(),
                            lastMappedFile.getFileFromOffset() + this.mappedFileSize,
                            this.mappedFileSize,
                            this.mappedFiles.size());
                    } else {
        
                        //因为RocketMQ定时删除存储文件
                        //所以第一个文件偏移开始并不一定是000000.
                        //同理可得offset / this.mappedFileSize并不能定位到具体文件
                        //所以还需要减去第一个文件的偏移/文件大小,算出磁盘中起始第几个文件
                        int index = (int) ((offset / this.mappedFileSize) - (firstMappedFile.getFileFromOffset() / this.mappedFileSize));
                        MappedFile targetFile = null;
                        try {
        
                            //获取映射文件
                            targetFile = this.mappedFiles.get(index);
                        } catch (Exception ignored) {
        
                        }
                        //再次检测是否在文件范围内
                        if (targetFile != null && offset >= targetFile.getFileFromOffset()
                            && offset < targetFile.getFileFromOffset() + this.mappedFileSize) {
        
                            return targetFile;
                        }
    
                        //遍历所有文件查找
                        for (MappedFile tmpMappedFile : this.mappedFiles) {
        
                            if (offset >= tmpMappedFile.getFileFromOffset()
                                && offset < tmpMappedFile.getFileFromOffset() + this.mappedFileSize) {
        
                                return tmpMappedFile;
                            }
                        }
                    }
    
                    //如果配置了没找到返回第一个,就返回第一个文件
                    if (returnFirstOnNotFound) {
        
                        return firstMappedFile;
                    }
                }
            } catch (Exception e) {
        
                log.error("findMappedFileByOffset Exception", e);
            }
    
            return null;
        }

## MappedFile 内存映射文件 ##

字段属性

//操作系统每页大小，默认4k 。
        public static final int OS_PAGE_SIZE = 1024 * 4;
        protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    
        //当前JVM 实例中MappedFile虚拟内存。
        private static final AtomicLong TOTAL_MAPPED_VIRTUAL_MEMORY = new AtomicLong(0);
    
        //当前JVM 实例中MappedFile 对象个数。
        private static final AtomicInteger TOTAL_MAPPED_FILES = new AtomicInteger(0);
        //当前该文件的写指针，从0 开始（内存映射文件中的写指针） 。
        protected final AtomicInteger wrotePosition = new AtomicInteger(0);
        //当前文件的提交指针，如果开启transientStore PoolEnable，则数据会存储在TransientStorePool 中，
        // 然后提交到内存映射ByteBuffer中，再刷写到磁盘。
        protected final AtomicInteger committedPosition = new AtomicInteger(0);
        //刷写到磁盘指针，该指针之前的数据持久化到磁盘中。
        private final AtomicInteger flushedPosition = new AtomicInteger(0);
        protected int fileSize;
        protected FileChannel fileChannel;
        /**
         * Message will put to here first, and then reput to FileChannel if writeBuffer is not null.
         */
        //堆内存ByteBuffer ， 如果不为空，数据首先将存储在该Buffer 中，
        // 然后提交到MappedFile 对应的内存映射文件Buffer 。
        // transientStorePoolEnable为true 时不为空。
        protected ByteBuffer writeBuffer = null;
        //堆内存池， transientStorePoolEnable 为true时启用。
        protected TransientStorePool transientStorePool = null;
        private String fileName;
        //该文件的初始偏移量。
        private long fileFromOffset;
        private File file;
        //物理文件对应的内存映射Buffer 。
        private MappedByteBuffer mappedByteBuffer;
        //文件最后一次内容写入时间。
        private volatile long storeTimestamp = 0;
        //是否是MappedFileQueue 队列中第一个文件。
        private boolean firstCreateInQueue = false;

### **初始化** ###

若transientStorePoolEnable 为true ，则初始化MappedFile 的writeBuffer .内存先存储堆外内存,然后通过Commit线程将数据提交到内存映射Buffer中,再通过Flush线程将内存映射Buffer 中的数据持久化到磁盘中.否则则将数据提交到文件映射通道中,然后在刷入磁盘

public void init(final String fileName, final int fileSize,
            final TransientStorePool transientStorePool) throws IOException {
        
            init(fileName, fileSize);
            //如果transientStorePoolEnable 为true ，则初始化MappedFile 的writeBuffer ，
            this.writeBuffer = transientStorePool.borrowBuffer();
            this.transientStorePool = transientStorePool;
        }
    
    
        private void init(final String fileName, final int fileSize) throws IOException {
        
            this.fileName = fileName;
            this.fileSize = fileSize;
            this.file = new File(fileName);
            //fileFromOffset为文件名
            this.fileFromOffset = Long.parseLong(this.file.getName());
            boolean ok = false;
    
            ensureDirOK(this.file.getParent());
    
            try {
        
                //创建文件读写通道
                this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
                //将文件映射内存中
                this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
                //重新计算MappedFile虚拟内存。
                TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
                //计算文件个数
                TOTAL_MAPPED_FILES.incrementAndGet();
                ok = true;
            } catch (FileNotFoundException e) {
        
                log.error("Failed to create file " + this.fileName, e);
                throw e;
            } catch (IOException e) {
        
                log.error("Failed to map file " + this.fileName, e);
                throw e;
            } finally {
        
                //执行失败,但通道创建成功,则关闭通道
                if (!ok && this.fileChannel != null) {
        
                    this.fileChannel.close();
                }
            }
        }

### 提交 ###

也就是将堆外数据写入channel

/**
         *  执行提交操作
         * @param commitLeastPages 本次提交最小的页数
         * @return
         */
        public int commit(final int commitLeastPages) {
        
            /*
            writeBuffer如果为空，直接返回wrotePosition 指针，无须执行commit 操作，
            表明commit 操作的实际是writeBuffer堆外内存
             */
            if (writeBuffer == null) {
        
                //no need to commit data to file channel, so just regard wrotePosition as committedPosition.
                return this.wrotePosition.get();
            }
            //判断是否执行commit 操作,主要判断页是否满足
            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.
            ///所有脏数据已经写入channel,且该文件已经提交满了.
            if (writeBuffer != null && this.transientStorePool != null && this.fileSize == this.committedPosition.get()) {
        
                //归还堆外内存给堆内存池
                this.transientStorePool.returnBuffer(writeBuffer);
                //释放GC
                this.writeBuffer = null;
            }
    
            //返回最新提交位置
            return this.committedPosition.get();
        }

是否执行commit 操作

//判断是否执行commit 操作
        protected boolean isAbleToCommit(final int commitLeastPages) {
        
            int flush = this.committedPosition.get();
            int write = this.wrotePosition.get();
    
            //若文件已满,返回true
            if (this.isFull()) {
        
                return true;
            }
    
            if (commitLeastPages > 0) {
        
                //当前writeBuffe 的写指针与上一次提交的指针（committedPosition)的差值
                // 除以OS_ PAGE_ SIZE 得到当前脏页的数量
                //如果大于commitLeastPages 则返回true
                return ((write / OS_PAGE_SIZE) - (flush / OS_PAGE_SIZE)) >= commitLeastPages;
            }
    
            //如果commitLeastPages 小于0 表示只要存在脏页就提交。
            return write > flush;
        }

具体的提交实现

protected void commit0(final int commitLeastPages) {
        
            int writePos = this.wrotePosition.get();
            int lastCommittedPosition = this.committedPosition.get();
    
            //说明有最新数据写,将writeBuffer中的数据提交到文件通道FileChannel 中。
            if (writePos - this.committedPosition.get() > 0) {
        
                try {
        
                    ByteBuffer byteBuffer = writeBuffer.slice();
                    byteBuffer.position(lastCommittedPosition);
                    byteBuffer.limit(writePos);
                    //设置当前文件提交指针位移
                    this.fileChannel.position(lastCommittedPosition);
                    //将writeBuffer数据提交到文件通道FileChannel 中。
                    this.fileChannel.write(byteBuffer);
                    //更新当前文件提交指针位移
                    this.committedPosition.set(writePos);
                } catch (Throwable e) {
        
                    log.error("Error occurred when commit data to FileChannel.", e);
                }
            }
        }

### 刷盘 ###

将内存数据刷入磁盘

/**
         * @return The current flushed position
         * 将内存中的数据刷写到磁盘
         */
        public int flush(final int flushLeastPages) {
        
            if (this.isAbleToFlush(flushLeastPages)) {
        
                if (this.hold()) {
        
                    //获取该文件内存映射写指针
                    //如果开启了堆内存池,则是堆内存写指针
                    int value = getReadPosition();
    
                    try {
        
                        //We only append data to fileChannel or mappedByteBuffer, never both.
                        //说明使用了堆内存,执行到这里堆内存已经写入fileChannel中了,重新刷入磁盘就行了
                        if (writeBuffer != null || this.fileChannel.position() != 0) {
        
                            //文件的所有待定修改立即同步到磁盘,布尔型参数表示在方法返回值前文件的元数据（metadata）是否也要被同步更新到磁盘
                            this.fileChannel.force(false);
                        } else {
        
                            //说明没用堆内存ByteBuffer,直接使用内存映射刷入即可.
                            this.mappedByteBuffer.force();
                        }
                    } catch (Throwable e) {
        
                        log.error("Error occurred when force data to disk.", e);
                    }
    
                    //记录刷入磁盘最新指针
                    this.flushedPosition.set(value);
                    this.release();
                } else {
        
                    log.warn("in flush, hold failed, flush offset = " + this.flushedPosition.get());
                    this.flushedPosition.set(getReadPosition());
                }
            }
            //返回刷入磁盘最新指针
            return this.getFlushedPosition();
        }

获取最大可读的位置

public int getReadPosition() {
        
            //若没开启堆内存临时存储
            return this.writeBuffer == null ?
                    //返回该文件的写位置
                    this.wrotePosition.get() :
                    //返回上次堆内存写入通道的位置
                    this.committedPosition.get();
        }

### 销毁 ###

/**
         * MappedFile文件销毁
         *
         * @param intervalForcibly  拒绝被销毁的最大存活时间。
         * @return
         */
        public boolean destroy(final long intervalForcibly) {
        
            //关闭MappedFile
            this.shutdown(intervalForcibly);
    
            //判断是否清理完成
            if (this.isCleanupOver()) {
        
                try {
        
                    //关闭通道
                    this.fileChannel.close();
                    log.info("close file channel " + this.fileName + " OK");
    
                    long beginTime = System.currentTimeMillis();
                    //删除整个物理文件
                    boolean result = this.file.delete();
                    log.info("delete file[REF:" + this.getRefCount() + "] " + this.fileName
                        + (result ? " OK, " : " Failed, ") + "W:" + this.getWrotePosition() + " M:"
                        + this.getFlushedPosition() + ", "
                        + UtilAll.computeElapsedTimeMilliseconds(beginTime));
                } catch (Exception e) {
        
                    log.warn("close file channel " + this.fileName + " Failed. ", e);
                }
    
                return true;
            } else {
        
                log.warn("destroy mapped file[REF:" + this.getRefCount() + "] " + this.fileName
                    + " Failed. cleanupOver: " + this.cleanupOver);
            }
    
            return false;
        }

关闭MappedFile

/**
         * 关闭MappedFile
         * @param intervalForcibly  拒绝被销毁的最大存活时间。
         */
        public void shutdown(final long intervalForcibly) {
        
            //默认true
            if (this.available) {
        
                //初次调用时available 为true ，设置available为fal se
                this.available = false;
                //设置初次关闭的时间戳
                this.firstShutdownTimestamp = System.currentTimeMillis();
                //释放资源,引用次数小于1 的情况下才会释放资源
                this.release();
            } else if (this.getRefCount() > 0) {
        
                //如果引用次数大于0
                //对比当前时间与firstShutdownTimestamp ，如果已经超过了其最大拒绝存活期，每执行
                //一次，将引用数减少1000 ，直到引用数小于0 时通过执行release方法释放资源。
                if ((System.currentTimeMillis() - this.firstShutdownTimestamp) >= intervalForcibly) {
        
                    this.refCount.set(-1000 - this.getRefCount());
                    this.release();
                }
            }
        }

减少引用并释放资源

public void release() {
        
            //引用减1
            long value = this.refCount.decrementAndGet();
            if (value > 0)
                return;
    
            synchronized (this) {
        
                //释放堆外内存
                this.cleanupOver = this.cleanup(value);
            }
        }

释放堆外内存

/**
         * 释放堆外内存
         */
        public boolean cleanup(final long currentRef) {
        
            //如果available为true ，表示MappedFile当前可用，无须清理，
            if (this.isAvailable()) {
        
                log.error("this file[REF:" + currentRef + "] " + this.fileName
                    + " have not shutdown, stop unmapping.");
                return false;
            }
    
            //如果资源已经被清除，返回true
            if (this.isCleanupOver()) {
        
                log.error("this file[REF:" + currentRef + "] " + this.fileName
                    + " have cleanup, do not do it again.");
                return true;
            }
    
            //如果是堆外内存，调用堆外内存的cleanup 方法清除
            clean(this.mappedByteBuffer);
            //维护虚拟内存
            TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(this.fileSize * (-1));
            //对象个数-1
            TOTAL_MAPPED_FILES.decrementAndGet();
            log.info("unmap file[REF:" + currentRef + "] " + this.fileName + " OK");
            return true;
        }

判断是否清理完成

/**
         * 判断是否清理完成
         * @return
         */
        public boolean isCleanupOver() {
        
            return
            //引用次数小于等于0
                    this.refCount.get() <= 0
                            //release 成功将MappedByteBuffer资源释放
                            && this.cleanupOver;
        }

### 其他方法 ###

/**
         *  查找pos 到当前最大可读之间的数据
         * @param pos
         * @return
         */
        public SelectMappedBufferResult selectMappedBuffer(int pos) {
        
            //获取最大可读数据位置
            int readPosition = getReadPosition();
            //若有数据可读
            if (pos < readPosition && pos >= 0) {
        
                if (this.hold()) {
        
                    /*
                    操作ByteBuffer 时如果使用了slice （） 方法，对其ByteBuffer 进行读取时一般手动指定
                    position 与limit 指针，而不是调用flip 方法来切换读写状态。
                     */
                    //由于在整个写入期间都未曾改变MappedByteBuffer的指针
                    //所以mappedByteBuffer.slice（）方法返回的共享缓存区空间为整个MappedFile
                    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;
        }

## TransientStorePool 临时堆外内存存储池 ##

RocketMQ 单独创建一个MappedByteBuffer 内存缓存池  
RokcetMQ 引人该机制主要的原因:提供一种内存锁定，将当前堆外内存一直锁定在内存中，避免被进程将内存交换到磁盘。

**字段**

//avaliableBuffers个数，可通过在broker 中配置文件中设置transientStorePoolSize， 默认为5 。
        private final int poolSize;
        //每个ByteBuffer 大小， 默认为mapedFileSizeCommitLog ，表明TransientStorePool 为commitlog 文件服务。
        private final int fileSize;
        //ByteBuffer 容器，双端队列。
        private final Deque<ByteBuffer> availableBuffers;
        //消息存储配置
        private final MessageStoreConfig storeConfig;

初始化

public void init() {
        
            //创建poolSize个堆外内存
            for (int i = 0; i < poolSize; i++) {
        
                //每个堆外内存大小fileSize
                ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize);
    
                final long address = ((DirectBuffer) byteBuffer).address();
                Pointer pointer = new Pointer(address);
                //利用com.sun.jna.Library 类库将该批内存锁定，避免被置换到交换区，提高存储性能。
                LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize));
    
                //添加双端队列
                availableBuffers.offer(byteBuffer);
            }
        }

# RocketMQ 存储文件 #

RocketMQ 存储路径为$｛ RocketMQ \_HOME\} / store

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]

1.  commitlog ：消息存储目录。
2.  config ：运行期间一些配置信息，主要包括下列信息。  
    consumerFilter.json ： 主题消息过滤信息。  
    consumerOffset.json ： 集群消费模式消息消费进度。  
    delayOffset.json：延时消息队列拉取进度。  
    subscriptionGroup.json ： 消息消费组配置信息。  
    topics.json:topic 配置属性。
3.  consumequeue ：消息消费队列存储目录。
4.  index ：消息索引文件存储目录。
5.  abort ：如果存在abort 文件说明Broker 非正常关闭，该文件默认启动时创建，正常退出之前删除。
6.  checkpoint ：文件检测点，存储commitlog 文件最后一次刷盘时间戳、consumequeue最后一次刷盘时间、index  
    索引文件最后一次刷盘时间戳。

## Commitlog 文件 ##

每条消息的前面4 个字节存储该条消息的总长度。  
![在这里插入图片描述][20200828183538744.png_pic_center]  
获取当前Commitlog 目录最小偏移量

//获取当前Commitlog 目录最小偏移量
        public long getMinOffset() {
        
            //获取第一个文件
            MappedFile mappedFile = this.mappedFileQueue.getFirstMappedFile();
            if (mappedFile != null) {
        
                //第一个文件可用
                if (mappedFile.isAvailable()) {
        
                    //返回该文件起始偏移量
                    return mappedFile.getFileFromOffset();
                } else {
        
                    //返回下个文件的起始偏移量
                    return this.rollNextFile(mappedFile.getFileFromOffset());
                }
            }
    
            return -1;
        }

根据该offset返回下一个文件的起始偏移量

public long rollNextFile(final long offset) {
        
            //获取一个文件大小,可配置,默认1G.
            int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
            //offset必须是mappedFileSize倍数
            //回到下个文件的起始偏移量
            return offset + mappedFileSize - offset % mappedFileSize;
        }

根据偏移量与消息长度查找消息

public SelectMappedBufferResult getMessage(final long offset, final int size) {
        
            //获取默认文件大小
            int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog();
            //首先根据偏移找到所在的物理偏移量
            MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, offset == 0);
            if (mappedFile != null) {
        
                //得到在文件内的偏移量
                int pos = (int) (offset % mappedFileSize);
                //从该偏移量读取size长度的内容返回
                return mappedFile.selectMappedBuffer(pos, size);
            }
            return null;
        }

## ConsumeQueue文件 ##

同一主题的消息不连续地存储在commitlog 文件中，如果消息消费者直接从消息存储文件（commitlog）中去遍历查找订阅主题下的消息，效率将极其低下，RocketMQ 为了适应消息消费的检索需求，设计了消息消费队列文件（Consumequeue），该文件可以看成是Commitlog 关于消息消费的“索引”文件  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]  
![在这里插入图片描述][20200828193523419.png_pic_center]  
单个ConsumeQueue 文件中默认包含30 万个条目，单个文件的长度为30w × 20 字节

根据startIndex获取消息消费队列条目

/**
         * 
         * @param startIndex  消息条目索引
         * @return
         */
        public SelectMappedBufferResult getIndexBuffer(final long startIndex) {
        
            int mappedFileSize = this.mappedFileSize;
            //得到在consumequeue中的物理偏移,每个队列条目占用20字节内存
            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;
        }

根据消息存储时间来查找该消息条目索引

/**
         * 
         * @param timestamp
         * @return
         */
        public long getOffsetInQueueByTime(final long timestamp) {
        
            //根据时间戳定位到物理文件
            MappedFile mappedFile = this.mappedFileQueue.getMappedFileByTime(timestamp);
            if (mappedFile != null) {
        
                long offset = 0;
                //计算最低位索引
                int low = minLogicOffset > mappedFile.getFileFromOffset() ? (int) (minLogicOffset - mappedFile.getFileFromOffset()) : 0;
                int high = 0;
                int midOffset = -1, targetOffset = -1, leftOffset = -1, rightOffset = -1;
                long leftIndexValue = -1L, rightIndexValue = -1L;
                //获取当前存储commitLog文件中有效的最小消息物理偏移量minPhysicOffset ，
                long minPhysicOffset = this.defaultMessageStore.getMinPhyOffset();
                SelectMappedBufferResult sbr = mappedFile.selectMappedBuffer(0);
                if (null != sbr) {
        
                    ByteBuffer byteBuffer = sbr.getByteBuffer();
                    //设置hign为文件最后一个元素
                    high = byteBuffer.limit() - CQ_STORE_UNIT_SIZE;
                    try {
        
                        while (high >= low) {
        
                            //获取中间偏移
                            midOffset = (low + high) / (2 * CQ_STORE_UNIT_SIZE) * CQ_STORE_UNIT_SIZE;
                            byteBuffer.position(midOffset);
                            //获取8字节该消息的物理偏移量
                            long phyOffset = byteBuffer.getLong();
                            //获取4字节该消息长度
                            int size = byteBuffer.getInt();
                            //小于当前的最小物理偏移量,说明消息无效,继续二分查找
                            if (phyOffset < minPhysicOffset) {
        
                                //从下一个消息条目开始二分查找
                                low = midOffset + CQ_STORE_UNIT_SIZE;
                                leftOffset = midOffset;
                                continue;
                            }
    
                            //根据消息偏移量和消息长度获取消息的存储时间戳。
                            long storeTime =
                                this.defaultMessageStore.getCommitLog().pickupStoreTimestamp(phyOffset, size);
                            //存储时间小于0
                            if (storeTime < 0) {
        
                                //消息为无效消息，直接返回0 。
                                return 0;
                            } else if (storeTime == timestamp) {
        
                                //如果存储时间戳等于待查找时间戳，说明查找到匹配消息，设置targetOffset 并跳 出循环。
                                targetOffset = midOffset;
                                break;
                            } else if (storeTime > timestamp) {
        
                                /*
                                //如果存储时间戳大于待查找时间戳，说明待查找信息小于midOffset ，则设置high
                                //为尾端消息-1个消息条目， 并设置rightlndexValue 等于midOffset 。
                                 */
                                high = midOffset - CQ_STORE_UNIT_SIZE;
                                rightOffset = midOffset;
                                rightIndexValue = storeTime;
                            } else {
        
                                /*
                                 如果存储时间小于待查找时间戳，说明待查找消息在大于midOffset ，则设置low
                                 为midOffset ，并设置leftlndexYalu巳等于mid Offset。
                                 */
                                low = midOffset + CQ_STORE_UNIT_SIZE;
                                leftOffset = midOffset;
                                leftIndexValue = storeTime;
                            }
                        }
    
                        //找到了存储时间戳等于待查找时间戳的消息
                        if (targetOffset != -1) {
        
                            offset = targetOffset;
                        } else {
        
                            //返回当前时间戳大并且最接近待查找的偏移量
                            if (leftIndexValue == -1) {
        
                                offset = rightOffset;
                            } else if (rightIndexValue == -1) {
        
                            //返回的消息比待查找时间戳小并且最接近查找的偏移量。
                                offset = leftOffset;
                            } else {
        
                                //比较2者,选择最靠近时间的偏移量
                                offset =
                                    Math.abs(timestamp - leftIndexValue) > Math.abs(timestamp
                                        - rightIndexValue) ? rightOffset : leftOffset;
                            }
                        }
    
                        //返回该消息条目索引
                        return (mappedFile.getFileFromOffset() + offset) / CQ_STORE_UNIT_SIZE;
                    } finally {
        
                        //释放引用
                        sbr.release();
                    }
                }
            }
            //说明找到的消息偏移小于该物理偏移量,字节返回
            return 0;
        }

根据当前偏移量获取下一个文件的起始偏移量

//
        public long rollNextFile(final long index) {
        
            int mappedFileSize = this.mappedFileSize;
            //获取一个文件包含多少个消息消费队列条目
            int totalUnitsInFile = mappedFileSize / CQ_STORE_UNIT_SIZE;
            //获取下一个文件的起始偏移量。
            return index + totalUnitsInFile - index % totalUnitsInFile;
        }

## Index ##

Hash 索引机制为消息建立索引,提高根据主题与消息队列检索消息的速度，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]

1.  IndexHeader 头部，包含40 个字节，记录该IndexFile 的统计信息，其结构如下。  
    beginTimestamp： 该索引文件中包含消息的最小存储时间。  
    endTimestamp ： 该索引文件中包含消息的最大存储时间。  
    beginPhyoffset ： 该索引文件中包含消息的最小物理偏移量（ commitlog 文件偏移量） 。  
    endPhyoffset ：该索引文件中包含消息的最大物理偏移量（ commitlog 文件偏移量） 。  
    hashslotCount:hashslot 个数，并不是hash 槽使用的个数，在这里意义不大。  
    indexCount: Index 条目列表当前已使用的个数， Index 条目在Index 条目列表中按顺序存储。
2.  Hash 槽， 一个IndexFile 默认包含500 万个Hash 槽，每个Hash 槽存储的是落在该 Hash槽的hashcode 最新的In dex 的索引。
3.  Index 条目列表，默认一个索引文件包含2000 万个条目，每一个Index 条目结构如下。  
    hashcode: key的hashcode 。  
    phyoffset ： 消息对应的物理偏移量。  
    timedif：该消息存储时间与第一条消息的时间戳的差值，小于0该消息无效。  
    prelndexNo ：该条目的前一条记录的Index 索引， 当出现hash 冲突时， 构建的链表结构。

字段

//每个槽的大小,4字节
        private static int hashSlotSize = 4;
        //每个Index 条目大小,20字节
        private static int indexSize = 20;
        //主要用来判断槽有没有被占用,0表没有
        private static int invalidIndex = 0;
        //一个IndexFile 默认包含500 万个Hash 槽，每个Hash 槽存储的是落在该Hash槽的hashcode 最新的Index 的索引。
        private final int hashSlotNum;
        //允许最大条目数
        private final int indexNum;
        //内存映射文件
        private final MappedFile mappedFile;
        //文件通道
        private final FileChannel fileChannel;
        //物理文件对应的内存映射Buffer 。
        private final MappedByteBuffer mappedByteBuffer;
        //IndexHeader头部
        private final IndexHeader indexHeader;

**IndexHeader字段属性**

//包含40个字节
        public static final int INDEX_HEADER_SIZE = 40;
        //该索引文件中包含消息的最小存储时间。
        private static int beginTimestampIndex = 0;
        // 该索引文件中包含消息的最大存储时间。
        private static int endTimestampIndex = 8;
        // 该索引文件中包含消息的最小物理偏移量（ commitlog 文件偏移量） 。
        private static int beginPhyoffsetIndex = 16;
        //该索引文件中包含消息的最大物理偏移量（ commitlog 文件偏移量）
        private static int endPhyoffsetIndex = 24;
        //hashslot 个数，并不是hash 槽使用的个数
        private static int hashSlotcountIndex = 32;
        private static int indexCountIndex = 36;
        private final ByteBuffer byteBuffer;
        private AtomicLong beginTimestamp = new AtomicLong(0);
        private AtomicLong endTimestamp = new AtomicLong(0);
        private AtomicLong beginPhyOffset = new AtomicLong(0);
        private AtomicLong endPhyOffset = new AtomicLong(0);
        //槽的使用个数
        private AtomicInteger hashSlotCount = new AtomicInteger(0);
    
        //Index 条目列表当前已使用的个数， Index 条目在Index 条目列表中按顺序存储。
        private AtomicInteger indexCount = new AtomicInteger(1);

将消息索引键与消息偏移量映射关系写入到IndexFile

/**
         * 将消息索引键与消息偏移量映射关系写入到IndexFile
         * @param key 消息索引
         * @param phyOffset 消息物理偏移量
         * @param storeTimestamp 消息存储时间
         * @return
         */
          public boolean putKey(final String key, final long phyOffset, final long storeTimestamp) {
        
            //如果当前已使用条目大于等于允许最大条目数时，则返回fasle ，表示当前索引文件已写满
            if (this.indexHeader.getIndexCount() < this.indexNum) {
        
                //根据key算出key 的hashcode
                int keyHash = indexKeyHashMethod(key);
                //定位hasbcode对应的hash槽下标
                int slotPos = keyHash % this.hashSlotNum;
                //hashcode对应的hash槽的物理地址=头部40字节+对应hash槽下标*槽大小
                int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
    
                FileLock fileLock = null;
    
                try {
        
    
                    // fileLock = this.fileChannel.lock(absSlotPos, hashSlotSize,
                    // false);
                    //读取hash 槽中存储的数据,每个槽占用4字节,也就是getInt就可以了
                    int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
    
                    //如果hash 槽存储的数据小于等于0 或大于当前索引文件中的索引条目格式，则将slotValue 设置为0
                    //说明槽尚未占用
                    if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()) {
        
                        slotValue = invalidIndex;
                    }
    
                    //计算待存储消息的时间戳与第一条消息时间戳的差值，并转换成秒。
                    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;
                    }
    
                    //计算新添加条目起始偏移量
                    int absIndexPos =
                            //头部字节长度
                        IndexHeader.INDEX_HEADER_SIZE +
                                //hash槽数量*单个槽大小
                                this.hashSlotNum * hashSlotSize
                                //当前index条目个数*单个条目大小
                            + this.indexHeader.getIndexCount() * indexSize;
    
                    //之所以只存储HashCode 而不存储具体的key ， 是为
                    //了将Index 条目设计为定长结构，才能方便地检索与定位条目。
                    this.mappedByteBuffer.putInt(absIndexPos, keyHash);
                    //消息对应的物理偏移量。
                    this.mappedByteBuffer.putLong(absIndexPos + 4, phyOffset);
                    //该消息存储时间与第一条消息的时间戳的差值，小于0 该消息无效
                    this.mappedByteBuffer.putInt(absIndexPos + 4 + 8, (int) timeDiff);
                    //当产生hash槽冲突时,Hash 槽中存储的是该Hash Code 所对应的最新的Index 条目的下标，
                    // 新的Index 条目的最后4 个字节存储该Hash Code 上一个条目的Index 下标。
                    //如果Hash 槽中存储的值为0 或大于当前lndexFile 最大条目数或小于－ 1，表示该Hash 槽当前并没有与之对应的Index 条目。
                    this.mappedByteBuffer.putInt(absIndexPos + 4 + 8 + 4, slotValue);
    
                    //当前hash槽的值存入MappedByteBuffer 中。将覆盖原先Hash 槽的值。
                    this.mappedByteBuffer.putInt(absSlotPos, this.indexHeader.getIndexCount());
    
                    //如果当前文件只包含一个条目,默认值为1
                    if (this.indexHeader.getIndexCount() <= 1) {
        
                        //更新BeginPhyOffset和BeginTimestamp
                        this.indexHeader.setBeginPhyOffset(phyOffset);
                        this.indexHeader.setBeginTimestamp(storeTimestamp);
                    }
    
                    //slotValue为0,说明新增hash槽使用
                    if (invalidIndex == slotValue) {
        
                        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;
        }

根据索引key 查找消息

/**
         * 根据索引key 查找消息的实现方法
         * @param phyOffsets 查找到的消息物理偏移量。
         * @param key 索引key 。
         * @param maxNum 本次查找最大消息条数。
         * @param begin 开始时间戳。
         * @param end 结束时间戳。
         * @param lock
         */
        public void selectPhyOffset(final List<Long> phyOffsets, final String key, final int maxNum,
            final long begin, final long end, boolean lock) {
        
            if (this.mappedFile.hold()) {
        
                //根据key算出hash
                int keyHash = indexKeyHashMethod(key);
                //根据hash获得hash槽下标
                int slotPos = keyHash % this.hashSlotNum;
                //根据槽+索引头大小获得hash槽地址
                int absSlotPos = IndexHeader.INDEX_HEADER_SIZE + slotPos * hashSlotSize;
    
                FileLock fileLock = null;
                try {
        
                    if (lock) {
        
                        // fileLock = this.fileChannel.lock(absSlotPos,
                        // hashSlotSize, true);
                    }
    
                    //获取槽记录的对应item条目下标
                    int slotValue = this.mappedByteBuffer.getInt(absSlotPos);
                    // if (fileLock != null) {
        
                    // fileLock.release();
                    // fileLock = null;
                    // }
    
                    //如果对应的Hash 槽中存储的数据小于1 或大于当前索引条目个数则表示该
                    //Hash Code 没有对应的条目， 直接返回。
                    if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCount()
                        || this.indexHeader.getIndexCount() <= 1) {
        
                    } else {
        
                        for (int nextIndexToRead = slotValue; ; ) {
        
                            //查找到的消息条目数量达到,则结束
                            if (phyOffsets.size() >= maxNum) {
        
                                break;
                            }
    
                            //获得item条目下标
                            int absIndexPos =
                                IndexHeader.INDEX_HEADER_SIZE + this.hashSlotNum * hashSlotSize
                                    + nextIndexToRead * indexSize;
    
                            int keyHashRead = this.mappedByteBuffer.getInt(absIndexPos);
                            long phyOffsetRead = this.mappedByteBuffer.getLong(absIndexPos + 4);
    
                            long timeDiff = (long) this.mappedByteBuffer.getInt(absIndexPos + 4 + 8);
                            //上一个条目的Index 下标
                            int prevIndexRead = this.mappedByteBuffer.getInt(absIndexPos + 4 + 8 + 4);
    
                            //如果存储的时间差小于0,该消息无效.则直接结束；
                            if (timeDiff < 0) {
        
                                break;
                            }
    
                            timeDiff *= 1000L;
    
                            //算出存储的时间
                            long timeRead = this.indexHeader.getBeginTimestamp() + timeDiff;
                            boolean timeMatched = (timeRead >= begin) && (timeRead <= end);
    
                            //如果hashcode 匹配并且
                            // 消息存储时间介于待查找时间start 、end 之间则将消息物理偏移量加入到phyOffsets ，
                            if (keyHash == keyHashRead && timeMatched) {
        
                                phyOffsets.add(phyOffsetRead);
                            }
    
                            //若上一个条目下标没有对应条目
                            if (prevIndexRead <= invalidIndex
                                || prevIndexRead > this.indexHeader.getIndexCount()
                                    //与当前遍历的位置一样
                                || prevIndexRead == nextIndexToRead ||
                                    //不满足时间范围
                                    timeRead < begin) {
        
                                //则查找结束
                                break;
                            }
    							 //索引大于等于l并且小于Index 条目数，继续查找
                            //下一个查找的条目地址
                            nextIndexToRead = prevIndexRead;
                        }
                    }
                } catch (Exception e) {
        
                    log.error("selectPhyOffset exception ", e);
                } finally {
        
                    if (fileLock != null) {
        
                        try {
        
                            fileLock.release();
                        } catch (IOException e) {
        
                            log.error("Failed to release the lock", e);
                        }
                    }
    
                    this.mappedFile.release();
                }
            }
        }

## checkpoint ##

记录Comitlog 、ConsumeQueue 、Index 文件的刷盘时间点， 文件  
固定长度为4k ，其中只用该文件的前面24 个字节  
!\[在这里插入图片描述\](https://img-blog.csdnimg.cn/20200828225958846.png\#pic\_center

public class StoreCheckpoint {
        
        private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
        private final RandomAccessFile randomAccessFile;
        private final FileChannel fileChannel;
        private final MappedByteBuffer mappedByteBuffer;
        //commitlog 文件刷盘时间点。
        private volatile long physicMsgTimestamp = 0;
        //消息消费队列文件刷盘时间点。
        private volatile long logicsMsgTimestamp = 0;
        //索引文件刷盘时间点。
        private volatile long indexMsgTimestamp = 0;
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/6a2b1483716b491eb0e84aa02d8b2e25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/c01cc14074b54b20af487f0708d5198e.png
[20200828183538744.png_pic_center]: /images/20221123/2ef766bf5c2941cea976a3e5c36ccaa1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/8a0eb2ec57a7410f82e16a46c20c4057.png
[20200828193523419.png_pic_center]: /images/20221123/abfbd01ea4834657ade351a635e31803.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/d3b27790410f42dfa1e6e58d878d5d31.png