【消息队列开发】 把消息写入文件

// 下列的逻辑, 并不仅仅是 Message, 其他的 Java 中的对象, 也是可以通过这样的逻辑进行序列化和反序列化的.
// 如果要想让这个对象能够序列化或者反序列化, 需要让这个类能够实现 Serializable 接口.
public class BinaryTool {
    // 把一个对象序列化成一个字节数组
    public static byte[] toBytes(Object object) throws IOException {
        // 这个流对象相当于一个变长的字节数组.
        // 就可以把 object 序列化的数据给逐渐的写入到 byteArrayOutputStream 中, 再统一转成 byte[]
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
            try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream)) {
                // 此处的 writeObject 就会把该对象进行序列化, 生成的二进制字节数据, 就会写入到
                // ObjectOutputStream 中.
                // 由于 ObjectOutputStream 又是关联到了 ByteArrayOutputStream,
                // 最终结果就写入到 ByteArrayOutputStream 里了
                objectOutputStream.writeObject(object);
            }
            // 这个操作就是把 byteArrayOutputStream 中持有的二进制数据取出来, 转成 byte[]
            return byteArrayOutputStream.toByteArray();
        }
    }
    // 把一个字节数组, 反序列化成一个对象
    public static Object fromBytes(byte[] data) throws IOException, ClassNotFoundException {
        Object object = null;
        try (ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(data)) {
            try (ObjectInputStream objectInputStream = new ObjectInputStream(byteArrayInputStream)) {
                // 此处的 readObject, 就是从 data 这个 byte[] 中读取数据并进行反序列化.
                object = objectInputStream.readObject();
            }
        }
        return object;
    }
}

/*
 * 自定义一个异常类. 如果是我们的 mq 的业务逻辑中, 出现的异常, 就抛出这个异常对象,
 * 同时在构造方法中指定出现异常的原因信息
 */
public class MqException extends Exception {
    public MqException(String reason) {
        super(reason);
    }
}

// 这个方法用来把一个新的消息, 放到队列对应的文件中.
// queue 表示要把消息写入的队列. message 则是要写的消息.
public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
    // 1. 检查一下当前要写入的队列对应的文件是否存在.
    if (!checkFilesExits(queue.getName())) {
        throw new MqException("[MessageFileManager] 队列对应的文件不存在! queueName=" + queue.getName());
    }
    // 2. 把 Message 对象, 进行序列化, 转成二进制的字节数组.
    byte[] messageBinary = BinaryTool.toBytes(message);
    synchronized (queue) {
        // 3. 先获取到当前的队列数据文件的长度, 用这个来计算出该 Message 对象的 offsetBeg 和 offsetEnd
        // 把新的 Message 数据, 写入到队列数据文件的末尾. 此时 Message 对象的 offsetBeg , 就是当前文件长度 + 4
        // offsetEnd 就是当前文件长度 + 4 + message 自身长度.
        File queueDataFile = new File(getQueueDataPath(queue.getName()));
        // 通过这个方法 queueDataFile.length() 就能获取到文件的长度. 单位字节.
        message.setOffsetBeg(queueDataFile.length() + 4);
        message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
        // 4. 写入消息到数据文件, 注意, 是追加写入到数据文件末尾.
        try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
            try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                // 接下来要先写当前消息的长度, 占据 4 个字节的~~
                dataOutputStream.writeInt(messageBinary.length);
                // 写入消息本体
                dataOutputStream.write(messageBinary);
            }
        }
        // 5. 更新消息统计文件
        Stat stat = readStat(queue.getName());
        stat.totalCount += 1;
        stat.validCount += 1;
        writeStat(queue.getName(), stat);
    }
}