计算机程序的思惟逻辑 (61) - 内存映射文件及其应用 - 实现一个简单的消息队列

时间  2019-11-08
标签 计算机 程序 思惟 逻辑 内存 映射 文件 及其 应用 实现 一个 简单 消息 队列 繁體版
原文   原文链接

本系列文章经补充和完善，已修订整理成书《Java编程的逻辑》（马俊昌著），由机械工业出版社华章分社出版，于2018年1月上市热销，读者好评如潮！各大网店和书店有售，欢迎购买：京东自营连接 html

本节介绍内存映射文件，内存映射文件不是Java引入的概念，而是操做系统提供的一种功能，大部分操做系统都支持。java

咱们先来介绍内存映射文件的基本概念，它是什么，能解决什么问题，而后咱们介绍如何在Java中使用，咱们会设计和实现一个简单的、持久化的、跨程序的消息队列来演示内存映射文件的应用。数据库

基本概念

所谓内存映射文件，就是将文件映射到内存，文件对应于内存中的一个字节数组，对文件的操做变为对这个字节数组的操做，而字节数组的操做直接映射到文件上。这种映射能够是映射文件所有区域，也能够是只映射一部分区域。编程

不过，这种映射是操做系统提供的一种假象，文件通常不会立刻加载到内存，操做系统只是记录下了这回事，当实际发生读写时，才会按需加载。操做系统通常是按页加载的，页能够理解为就是一块，页的大小与操做系统和硬件相关，典型的配置多是4K, 8K等，当操做系统发现读写区域不在内存时，就会加载该区域对应的一个页到内存。数组

这种按需加载的方式，使得内存映射文件能够方便处理很是大的文件，内存放不下整个文件也没关系，操做系统会自动进行处理，将须要的内容读到内存，将修改的内容保存到硬盘，将再也不使用的内存释放。微信

在应用程序写的时候，它写的是内存中的字节数组，这个内容何时同步到文件上呢？这个时机是不肯定的，由操做系统决定，不过，只要操做系统不崩溃，操做系统会保证同步到文件上，即便映射这个文件的应用程序已经退出了。并发

在通常的文件读写中，会有两次数据拷贝，一次是从硬盘拷贝到操做系统内核，另外一次是从操做系统内核拷贝到用户态的应用程序。而在内存映射文件中，通常状况下，只有一次拷贝，且内存分配在操做系统内核，应用程序访问的就是操做系统的内核内存空间，这显然要比普通的读写效率更高。app

内存映射文件的另外一个重要特色是，它能够被多个不一样的应用程序共享，多个程序能够映射同一个文件，映射到同一块内存区域，一个程序对内存的修改，可让其余程序也看到，这使得它特别适合用于不一样应用程序之间的通讯。dom

操做系统自身在加载可执行文件的时候，通常都利用了内存映射文件，好比：spa

  • 按需加载代码，只有当前运行的代码在内存，其余暂时用不到的代码还在硬盘
  • 同时启动屡次同一个可执行文件，文件代码在内存也只有一份
  • 不一样应用程序共享的动态连接库代码在内存也只有一份

内存映射文件也有局限性，好比，它不太适合处理小文件，它是按页分配内存的，对于小文件，会浪费空间，另外，映射文件要消耗必定的操做系统资源，初始化比较慢。

简单总结下，对于通常的文件读写不须要使用内存映射文件，但若是处理的是大文件，要求极高的读写效率，好比数据库系统，或者须要在不一样程序间进行共享和通讯，那就能够考虑内存映射文件。

理解了内存映射文件的基本概念，接下来，咱们看怎么在Java中使用它。

用法

映射文件

内存映射文件须要经过FileInputStream/FileOutputStream或RandomAccessFile，它们都有一个方法：

public FileChannel getChannel() 复制代码

FileChannel有以下方法：

public MappedByteBuffer map(MapMode mode, long position, long size) throws IOException 复制代码

map方法将当前文件映射到内存，映射的结果就是一个MappedByteBuffer对象，它表明内存中的字节数组，待会咱们再来详细看它。map有三个参数，mode表示映射模式，positon表示映射的起始位置，size表示长度。

mode有三个取值：

  • MapMode.READ_ONLY：只读
  • MapMode.READ_WRITE：既读也写
  • MapMode.PRIVATE：私有模式，更改不反映到文件，也不被其余程序看到

这个模式受限于背后的流或RandomAccessFile，好比，对于FileInputStream，或者RandomAccessFile但打开模式是"r"，那mode就不能设为MapMode.READ_WRITE，不然会抛出异常。

若是映射的区域超过了现有文件的范围，则文件会自动扩展，扩展出的区域字节内容为0。

映射完成后，文件就能够关闭了，后续对文件的读写能够经过MappedByteBuffer。

看段代码，好比以读写模式映射文件"abc.dat"，代码能够为：

RandomAccessFile file = new RandomAccessFile("abc.dat","rw");
try {
    MappedByteBuffer buf = file.getChannel().map(MapMode.READ_WRITE, 0, file.length());
    //使用buf...
} catch (IOException e) {
    e.printStackTrace();
}finally{
    file.close();
}
复制代码

MappedByteBuffer

怎么来使用MappedByteBuffer呢？它是ByteBuffer的子类，而ByteBuffer是Buffer的子类。ByteBuffer和Buffer不仅是给内存映射文件提供的，它们是Java NIO中操做数据的一种方式，用于不少地方，方法也比较多，咱们只介绍一些主要相关的。

ByteBuffer能够简单理解为就是封装了一个字节数组，这个字节数组的长度是不可变的，在内存映射文件中，这个长度由map方法中的参数size决定。

ByteBuffer有一个基本属性position，表示当前读写位置，这个位置能够改变，相关方法是：

//获取当前读写位置
public final int position() //修改当前读写位置 public final Buffer position(int newPosition) 复制代码

ByteBuffer中有不少基于当前位置读写数据的方法，如：

//从当前位置获取一个字节
public abstract byte get();
//从当前位置拷贝dst.length长度的字节到dst
public ByteBuffer get(byte[] dst) //从当前位置读取一个int public abstract int getInt();
//从当前位置读取一个double
public abstract double getDouble();
//将字节数组src写入当前位置
public final ByteBuffer put(byte[] src) //将long类型的value写入当前位置 public abstract ByteBuffer putLong(long value);
复制代码

这些方法在读写后，都会自动增长position。

与这些方法相对应的，还有一组方法，能够在参数中直接指定position，好比：

//从index处读取一个int
public abstract int getInt(int index);
//从index处读取一个double
public abstract double getDouble(int index);
//在index处写入一个double
public abstract ByteBuffer putDouble(int index, double value);
//在index处写入一个long
public abstract ByteBuffer putLong(int index, long value);
复制代码

这些方法在读写时，不会改变当前读写位置position。

MappedByteBuffer本身还定义了一些方法：

//检查文件内容是否真实加载到了内存，这个值是一个参考值，不必定精确
public final boolean isLoaded() //尽可能将文件内容加载到内存 public final MappedByteBuffer load() //将对内存的修改强制同步到硬盘上 public final MappedByteBuffer force() 复制代码

消息队列

了解了内存映射文件的用法，接下来，咱们来看怎么用它设计和实现一个简单的消息队列，咱们称之为BasicQueue。

功能

BasicQueue是一个先进先出的循环队列，长度固定，接口主要是出队和入队，与以前介绍的容器类的区别是：

  • 消息持久化保存在文件中，重启程序消息不会丢失
  • 能够供不一样的程序进行协做，典型场景是，有两个不一样的程序，一个是生产者，另外一个是消费者，生成者只将消息放入队列，而消费者只从队列中取消息，两个程序经过队列进行协做，这种协做方式更灵活，相互依赖性小，是一种常见的协做方式。

BasicQueue的构造方法是：

public BasicQueue(String path, String queueName) throws IOException 复制代码

path表示队列所在的目录，必须已存在，queueName表示队列名，BasicQueue会使用以queueName开头的两个文件来保存队列信息，一个后缀是.data，保存实际的消息，另外一个后缀是.meta，保存元数据信息，若是这两个文件存在，则会使用已有的队列，不然会创建新队列。

BasicQueue主要提供两个方法，出队和入队，以下所示：

//入队
public void enqueue(byte[] data) throws IOException //出队 public byte[] dequeue() throws IOException 复制代码

与上节介绍的BasicDB相似，消息格式也是byte数组。BasicQueue的队列长度是有限的，若是满了，调用enqueue会抛出异常，消息的最大长度也是有限的，不能超过1020，若是超了，也会抛出异常。若是队列为空，dequeue返回null。

用法示例

BasicQueue的典型用法是生产者和消费者之间的协做，咱们来看下简单的示例代码。生产者程序向队列上放消息，每放一条，就随机休息一下子，代码为：

public class Producer {
    public static void main(String[] args) throws InterruptedException {
        try {
            BasicQueue queue = new BasicQueue("./", "task");
            int i = 0;
            Random rnd = new Random();
            while (true) {
                String msg = new String("task " + (i++));
                queue.enqueue(msg.getBytes("UTF-8"));
                System.out.println("produce: " + msg);
                Thread.sleep(rnd.nextInt(1000));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
复制代码

消费者程序从队列中取消息，若是队列为空，也随机睡一下子，代码为：

public class Consumer {
    public static void main(String[] args) throws InterruptedException {
        try {
            BasicQueue queue = new BasicQueue("./", "task");
            Random rnd = new Random();
            while (true) {
                byte[] bytes = queue.dequeue();
                if (bytes == null) {
                    Thread.sleep(rnd.nextInt(1000));
                    continue;
                }
                System.out.println("consume: " + new String(bytes, "UTF-8"));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
复制代码

假定这两个程序的当前目录同样，它们会使用一样的队列"task"。同时运行这两个程序，会看到它们的输出交替出现。

设计

咱们采用以下简单方式来设计BasicQueue：

  • 使用两个文件来保存消息队列，一个为数据文件，后缀为.data，一个是元数据文件.meta。
  • 在.data文件中使用固定长度存储每条信息，长度为1024，前4个字节为实际长度，后面是实际内容，每条消息的最大长度不能超过1020。
  • 在.meta文件中保存队列头和尾，指向.data文件中的位置，初始都是0，入队增长尾，出队增长头，到结尾时，再从0开始，模拟循环队列。
  • 为了区分队列满和空的状态，始终留一个位置不保存数据，当队列头和尾同样的时候表示队列为空，当队列尾的下一个位置是队列头的时候表示队列满。

基本设计以下图所示：

为简化起见，咱们暂不考虑因为并发访问等引发的一致性问题。

实现消息队列

下面来看BasicQueue的具体实现代码。

常量定义

BasicQueue中定义了以下常量，名称和含义以下：

// 队列最多消息个数，实际个数还会减1
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;
// 队列元数据文件大小 (head + tail)
private static final int META_SIZE = 8;
复制代码

内部组成

BasicQueue的内部成员主要就是两个MappedByteBuffer，分别表示数据和元数据：

private MappedByteBuffer dataBuf;
private MappedByteBuffer metaBuf; 
复制代码

构造方法

BasicQueue的构造方法代码是：

public BasicQueue(String path, String queueName) throws IOException {
    if (path.endsWith(File.separator)) {
        path += File.separator;
    }
    RandomAccessFile dataFile = null;
    RandomAccessFile metaFile = null;
    try {
        dataFile = new RandomAccessFile(path + queueName + ".data", "rw");
        metaFile = new RandomAccessFile(path + queueName + ".meta", "rw");

        dataBuf = dataFile.getChannel().map(MapMode.READ_WRITE, 0,
                DATA_FILE_SIZE);
        metaBuf = metaFile.getChannel().map(MapMode.READ_WRITE, 0,
                META_SIZE);
    } finally {
        if (dataFile != null) {
            dataFile.close();
        }
        if (metaFile != null) {
            metaFile.close();
        }
    }
}
复制代码

辅助方法

为了方便访问和修改队列头尾指针，咱们有以下方法：

private int head() {
    return metaBuf.getInt(0);
}

private void head(int newHead) {
    metaBuf.putInt(0, newHead);
}

private int tail() {
    return metaBuf.getInt(4);
}

private void tail(int newTail) {
    metaBuf.putInt(4, newTail);
}
复制代码

为了便于判断队列是空仍是满，咱们有以下方法：

private boolean isEmpty(){
    return head() == tail();
}

private boolean isFull(){
    return ((tail() + MSG_SIZE) % DATA_FILE_SIZE) == head();
}
复制代码

入队

代码为：

public void enqueue(byte[] data) throws IOException {
    if (data.length > MAX_MSG_BODY_SIZE) {
        throw new IllegalArgumentException("msg size is " + data.length
                + ", while maximum allowed length is " + MAX_MSG_BODY_SIZE);
    }
    if (isFull()) {
        throw new IllegalStateException("queue is full");
    }
    int tail = tail();
    dataBuf.position(tail);
    dataBuf.putInt(data.length);
    dataBuf.put(data);

    if (tail + MSG_SIZE >= DATA_FILE_SIZE) {
        tail(0);
    } else {
        tail(tail + MSG_SIZE);
    }
}
复制代码

基本逻辑是：

  1. 若是消息太长或队列满，抛出异常。
  2. 找到队列尾，定位到队列尾，写消息长度，写实际数据。
  3. 更新队列尾指针，若是已到文件尾，再从头开始。

出队

代码为：

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);

    if (head + MSG_SIZE >= DATA_FILE_SIZE) {
        head(0);
    } else {
        head(head + MSG_SIZE);
    }
    return data;
}
复制代码

基本逻辑是：

  1. 若是队列为空，返回null。
  2. 找到队列头，定位到队列头，读消息长度，读实际数据。
  3. 更新队列头指针，若是已到文件尾，再从头开始。
  4. 最后返回实际数据

小结

本节介绍了内存映射文件的基本概念及在Java中的的用法，在平常普通的文件读写中，咱们用到的比较少，但在一些系统程序中，它倒是常常被用到的一把利器，能够高效的读写大文件，且能实现不一样程序间的共享和通讯。

利用内存映射文件，咱们设计和实现了一个简单的消息队列，消息能够持久化，能够实现跨程序的生产者/消费者通讯，咱们演示了这个消息队列的功能、用法、设计和实现代码。

前面几节，咱们屡次提到过序列化的概念，它究竟是什么呢？

未完待续，查看最新文章，敬请关注微信公众号“老马说编程”(扫描下方二维码)，深刻浅出，老马和你一块儿探索Java编程及计算机技术的本质。用心原创，保留全部版权。

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程