【消息队列开发】 实现消息持久化

// 定义一个内部类, 来表示该队列的统计信息
// 有限考虑使用 static, 静态内部类.
static public class Stat {
    // 此处直接定义成 public, 就不再搞 get set 方法了.
    // 对于这样的简单的类, 就直接使用成员, 类似于 C 的结构体了.
    public int totalCount;  // 总消息数量
    public int validCount;  // 有效消息数量
}
// 预定消息文件所在的目录和文件名
// 这个方法, 用来获取到指定队列对应的消息文件所在路
private String getQueueDir(String queuename) {
    return ".data" + queuename;
}
// 这个方法用来获取该队列的消息数据文件路径
// 注意, 二进制文件, 使用 txt 作为后缀, 不太合适. txt 一般表示文本. 此处咱们也就不改.
// .bin / .dat
private String getQueueDataPath(String queueName) {
    return getQueueDir(queueName) + "/queue_data.txt";
}
// 这个方法用来获取该队列的消息统计文件路径
private String getQueueStatPath(String queueName) {
    return getQueueDir(queueName) + "/queue_stat.txt";
}
private Stat readStat(String queuename) {
    // 由于当前的消息统计文件是文本文件, 可以直接使用 Scanner 来读取文件内容
    Stat stat = new Stat();
    try(InputStream inputStream = new FileInputStream(getQueueStatPath(queuename))) {
        Scanner scanner = new Scanner(inputStream);
        stat.totalCount = scanner.nextInt();
        stat.validCount = scanner.nextInt();
        return stat;
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}
private void writeStat(String queueName, Stat stat) {
    // 使用 PrintWrite 来写文件.
    // OutputStream 打开文件, 默认情况下, 会直接把原文件清空. 此时相当于新的数据覆盖了旧的.
    try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))) {
        PrintWriter printWriter = new PrintWriter(outputStream);
        printWriter.print(stat.totalCount + "/t" + stat.validCount);
        printWriter.flush();
    } catch (IOException e) {
        e.printStackTrace();
    }
}
// 创建队列对应的文件和目录
public void createQueueFiles(String queueName) throws IOException {
    // 1. 先创建队列对应的消息目录
    File baseDir = new File(getQueueDir(queueName));
    if (!baseDir.exists()) {
        // 不存在, 就创建这个目录
        boolean ok = baseDir.mkdirs();
        if (!ok) {
            throw new IOException("创建目录失败! baseDir=" + baseDir.getAbsolutePath());
        }
    }
    // 2. 创建队列数据文件
    File queueDataFile = new File(getQueueDataPath(queueName));
    if (!queueDataFile.exists()) {
        boolean ok = queueDataFile.createNewFile();
        if (!ok) {
            throw new IOException("创建文件失败! queueDataFile=" + queueDataFile.getAbsolutePath());
        }
    }
    // 3. 创建消息统计文件
    File queueStatFile = new File(getQueueStatPath(queueName));
    if (!queueStatFile.exists()) {
        boolean ok = queueStatFile.createNewFile();
        if (!ok) {
            throw new IOException("创建文件失败! queueStatFile=" + queueStatFile.getAbsolutePath());
        }
    }
    // 4. 给消息统计文件, 设定初始值. 0\t0
    Stat stat = new Stat();
    stat.totalCount = 0;
    stat.validCount = 0;
    writeStat(queueName, stat);
}

public void destroyQueueFiles(String queueName) throws IOException {
    // 先删除里面的文件, 再删除目录.
    File queueDataFile = new File(getQueueDataPath(queueName));
    boolean ok1 = queueDataFile.delete();
    File queueStatFile = new File(getQueueStatPath(queueName));
    boolean ok2 = queueStatFile.delete();
    File baseDir = new File(getQueueDir(queueName));
    boolean ok3 = baseDir.delete();
    if (!ok1 || !ok2 || !ok3) {
        // 有任意一个删除失败, 都算整体删除失败.
        throw new IOException("删除队列目录和文件失败! baseDir=" + baseDir.getAbsolutePath());
    }
}
// 检查队列的目录和文件是否存在.
// 比如后续有生产者给 broker server 生产消息了, 这个消息就可能需要记录到文件上(取决于消息是否要持久化)
public boolean checkFilesExits(String queueName) {
    // 判定队列的数据文件和统计文件是否都存在!!
    File queueDataFile = new File(getQueueDataPath(queueName));
    if (!queueDataFile.exists()) {
        return false;
    }
    File queueStatFile = new File(getQueueStatPath(queueName));
    if (!queueStatFile.exists()) {
        return false;
    }
    return true;
}