RocketMQ-创建MappedFile本地文件

Posted 2020-09-18 瓜子

了解RocketMQ的都知道，它会保存所有的消息到本地文件。这个文件就是 MappedFile,每一个文件对应一个MappedFile.默认情况下大小位1g。

在MessageStoreConfig中的mapedFileSizeCommitLog设置，当然一半情况下是通过配置文件来设置的。文件路劲也都是在这个配置类里。

文件名格式是20位的数字，在这个类里生成，类似这样的（00000001000000000000 00100000000000000000 09000000000020000000）：

MappedFile由MappedFileQueue管理，下面是用来生成文件的方法：

   /**
     * 获取最后一个 MappedFile，若不存在或文件已满，则进行创建。
     * @param startOffset
     * @param needCreate
     * @return
     */
    public MapedFile getLastMapedFile(final long startOffset, boolean needCreate) {
        // 这个offeset是所有MappedFile的全局offset
        long createOffset = -1; // 创建文件开始offset。-1时，不创建
        MapedFile mapedFileLast = null;
        {
            this.readWriteLock.readLock().lock();
            if (this.mapedFiles.isEmpty()) {    // 一个映射文件都不存在
                createOffset = startOffset - (startOffset % this.mapedFileSize);
            } else {
                // 如果存在MappedFile对象，则获取最后一个List<MapedFile> mapedFiles = new ArrayList<MapedFile>()
                mapedFileLast = this.mapedFiles.get(this.mapedFiles.size() - 1);
            }
            this.readWriteLock.readLock().unlock();
        }

        if (mapedFileLast != null && mapedFileLast.isFull()) {  // 最后一个文件已满
            // 通过这里你可以知道这个offset是所有文件的offse，而不是一个mappedfile的offset
            createOffset = mapedFileLast.getFileFromOffset() + this.mapedFileSize;
        }

        // 创建文件
        if (createOffset != -1 && needCreate) {
            // 计算文件名。从此处我们可 以得知，MappedFile的文件命名规则：
            // 00000001000000000000 00100000000000000000 09000000000020000000二十位
            // fileName[n] = fileName[n - 1] + n * mappedFileSize fileName[0] = startOffset - (startOffset % this.mappedFileSize)
            String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
            log.info("shang\'s log >>> create mappedFile nextFilePath:{}",nextFilePath);
            String nextNextFilePath =
                    this.storePath + File.separator
                            + UtilAll.offset2FileName(createOffset + this.mapedFileSize);
            //  下一个文件的地址都算出来了，也对，就新建的时候可以直接加一个mapedfilesize来计算，就是1g。1024*1024*1024
            log.info("shang\'s log >>> nextNextFilePath:{}",nextNextFilePath);

            MapedFile mapedFile = null;

            // 两种方式创建文件
            if (this.allocateMapedFileService != null) {
                mapedFile =
                        this.allocateMapedFileService.putRequestAndReturnMapedFile(nextFilePath,
                                nextNextFilePath, this.mapedFileSize);
            } else {
                try {
                    mapedFile = new MapedFile(nextFilePath, this.mapedFileSize);
                } catch (IOException e) {
                    log.error("create mapedfile exception", e);
                }
            }

            if (mapedFile != null) {
                this.readWriteLock.writeLock().lock();
                if (this.mapedFiles.isEmpty()) {
                    mapedFile.setFirstCreateInQueue(true);
                }
                this.mapedFiles.add(mapedFile);
                this.readWriteLock.writeLock().unlock();
            }

            return mapedFile;
        }

        return mapedFileLast;
    }

有两种方式生成mappedfile文件，第一种可以参考这篇文章：http://www.cnblogs.com/guazi/p/6850988.html

最终都是通过标黄代码那样来生成文件的。我们来看一下，就是MappedFile的构造方法：

    // 创建mappedfile文件
    public MapedFile(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 {
　　　　　　　// 这两个都是与文件对应的mappedfile对象属性变量　
            this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
            this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
　　　　　　  // 这两个是rocketmq系统的共享静态变量，用来标示虚拟内存使用情况和文件数量
            TotalMapedVitualMemory.addAndGet(fileSize);
            TotalMapedFiles.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();
            }
        }
    }