您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 内存映射文件实战：MappedByteBuffer 原理与跨进程持久化消息队列实现

編輯：Java 綜合教學

核心概念：什么是内存映射文件？

内存映射文件将磁盘文件直接映射到进程虚拟内存区域，文件内容对应一个字节数组。**操作系统按需分页加载**（典型页大小 4KB/8KB），访问未加载区域时触发缺页中断自动读入。

关键特性

大文件友好：无需一次性读入内存，OS 自动管理换入换出。
零拷贝优势：传统 IO 需「磁盘→内核缓冲区→用户缓冲区」两次拷贝；MMF 仅需「磁盘→内核缓冲区（即用户映射区）」一次映射，用户态直接访问内核缓冲区。
进程间共享：多进程映射同一文件，共享同一物理内存页，一处修改、处处可见，天然适合 IPC。
持久化保证：OS 保证最终刷盘，即使进程崩溃数据不丢（可调用 force() 强制同步）。

Java API 核心用法

1. 创建映射

RandomAccessFile file = new RandomAccessFile("abc.dat", "rw");
MappedByteBuffer buf = file.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, file.length());
file.close(); // 映射建立后文件句柄可关闭

MapMode：READ_ONLY、READ_WRITE、PRIVATE（写时复制，不反映到文件、不共享）。

2. MappedByteBuffer 核心操作

继承 ByteBuffer，封装定长字节数组，维护 position 游标。

相对读写：get()、put(byte[])、getInt()、putLong() 等，自动推进 position。
绝对读写：getInt(index)、putLong(index, val)，不改变 position。
特有方法：isLoaded()、load()（预加载）、force()（强制刷盘）。

实战：跨进程持久化消息队列 BasicQueue

设计目标

FIFO 循环队列，固定容量。
消息持久化到文件，重启不丢失。
生产者/消费者跨进程协作（共享映射文件）。

存储结构设计

数据文件 .data：定长记录，每条 1024 字节（4 字节长度 + 1020 字节体）。
元数据文件 .meta：8 字节（4 字节 head 指针 + 4 字节 tail 指针），记录队列首尾在数据文件中的偏移量。
空一格判满/空：head == tail 为空；(tail + MSG_SIZE) % DATA_SIZE == head 为满。

队列结构示意图

核心常量

private static final int MAX_MSG_NUM = 1020 * 1024;     // 最大消息数
private static final int MAX_MSG_BODY_SIZE = 1020;      // 消息体最大长度
private static final int MSG_SIZE = MAX_MSG_BODY_SIZE + 4; // 单条占用
private static final int DATA_FILE_SIZE = MAX_MSG_NUM * MSG_SIZE; // ~1GB
private static final int META_SIZE = 8;                 // head + tail

构造方法：建立双映射

public BasicQueue(String path, String queueName) throws IOException {
    // 规范化路径
    RandomAccessFile dataFile = new RandomAccessFile(path + queueName + ".data", "rw");
    RandomAccessFile metaFile = new RandomAccessFile(path + queueName + ".meta", "rw");
    try {
        dataBuf = dataFile.getChannel().map(MapMode.READ_WRITE, 0, DATA_FILE_SIZE);
        metaBuf = metaFile.getChannel().map(MapMode.READ_WRITE, 0, META_SIZE);
    } finally {
        // 映射建立后关闭文件句柄不影响映射
        dataFile.close(); metaFile.close();
    }
}

入队

public void enqueue(byte[] data) throws IOException {
    if (data.length > MAX_MSG_BODY_SIZE) throw new IllegalArgumentException("消息过大");
    if (isFull()) throw new IllegalStateException("队列已满");

    int tail = tail();
    dataBuf.position(tail);
    dataBuf.putInt(data.length);
    dataBuf.put(data);

    // 循环更新 tail
    tail(tail + MSG_SIZE >= DATA_FILE_SIZE ? 0 : tail + MSG_SIZE);
}

出队

public byte[] dequeue() throws IOException {
    if (isEmpty()) return null;

    int head = head();
    dataBuf.position(head);
    int length = dataBuf.getInt();
    byte[] data = new byte[length];
    dataBuf.get(data);

    head(head + MSG_SIZE >= DATA_FILE_SIZE ? 0 : head + MSG_SIZE);
    return data;
}

生产者/消费者 Demo

// Producer.java
BasicQueue queue = new BasicQueue("./", "task");
while (true) {
    queue.enqueue(("task " + i++).getBytes("UTF-8"));
    Thread.sleep(random.nextInt(1000));
}

// Consumer.java
BasicQueue queue = new BasicQueue("./", "task");
while (true) {
    byte[] bytes = queue.dequeue();
    if (bytes == null) { Thread.sleep(random.nextInt(1000)); continue; }
    System.out.println("consume: " + new String(bytes, "UTF-8"));
}

两个独立 JVM 进程同时运行，通过共享 task.data/task.meta 实现零依赖协作。