用 Chronicle Queue 搭建低延迟本地消息队列：写入、读取与滚动配置

Chronicle Queue 是一个基于内存映射文件的持久化消息队列，适合对延迟敏感、又需要落盘与重放能力的单机场景，例如交易系统、行情分发、实时分析与进程间通信。它把消息写进磁盘上的队列文件，读取时通过内存映射访问，因此不需要网络往返，也不依赖外部 broker。这篇教程面向第一次接触 Chronicle Queue 的 Java 开发者，从加依赖开始，一步步写出可运行的生产者与消费者，再补上持久化、滚动策略、请求响应与流式读取等常见用法，最后说明它适合什么、不适合什么。如果你只是想快速查一下这个工具的基本定位，可以先看 Chronicle 的条目，再回到本文动手实践。

需要提前说明的是：本文给出的版本号、参数与代码结构来自公开的入门示例，实际项目请以官网当前发布的版本与文档为准；价格、许可与功能范围同样以官网当前信息为准。

准备工作

环境与依赖

Chronicle Queue 是 JVM 上的库，不需要单独安装服务端，也不需要启动 broker。你只需要一个可用的 JDK 与构建工具（Maven 或 Gradle），以及一块可写的本地磁盘目录用来存放队列文件。

Maven 项目在 pom.xml 中加入依赖：

<dependency>
  <groupId>net.openhft</groupId>
  <artifactId>chronicle-queue</artifactId>
  <version>5.24.0</version>
</dependency>

Gradle 项目则写成：

implementation 'net.openhft:chronicle-queue:5.24.0'

版本号请按官网当前发布版本替换，不要直接照抄示例中的数字。

需要先建立的几个概念

Chronicle Queue 的核心单位叫 excerpt，可以理解成一条自包含的记录。围绕它有三个角色：

• ExcerptAppender：向队列写入数据的写入端，通常每个线程持有一个。
• ExcerptReader：从队列读取数据的读取端，适合从头读取或按需遍历。
• ExcerptTailer：从队列尾部跟随读取，行为更接近流式消费者，只关心新到达的消息。

队列本身由 ChronicleQueue 对象表示，通过 builder 指定存储路径、滚动策略、文件大小上限等参数。队列文件按滚动周期切分成多个文件，读取时可以跨文件连续读。

目录与权限

队列路径必须指向一个真实存在或可被创建的目录，进程需要对它拥有读写权限。示例中常用 /tmp/my-queue 这类临时目录做试验，生产环境应换成独立的数据盘路径，例如 /data/persistent-queue，避免被系统清理或与其他应用争抢 IO。

操作步骤

第一步：创建队列并写入第一条消息

下面是一个最小可运行的生产者—消费者示例。它创建磁盘上的队列，写入一条带事件名的文本消息，然后读回来打印。

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.ExcerptReader;
import java.nio.file.Path;

public class SimpleQueueExample {
    public static void main(String[] args) {
        ChronicleQueue queue = ChronicleQueue.builder()
                .path(Path.of("/tmp/my-queue"))
                .build();

        ExcerptAppender appender = queue.acquireAppender();
        appender.writeDocument(w -> w
                .writeEventName("message")
                .writeText("Hello, Chronicle!"));

        try (ExcerptReader reader = queue.createReader()) {
            reader.readDocument(r -> {
                String message = r.readEventName();
                System.out.println("Received: " + message);
            });
        }

        queue.close();
    }
}

几个要点：

• writeDocument 接收一个写入闭包，闭包内的链式调用决定这条 excerpt 里有哪些字段。
• writeEventName 用来标记消息类型，读取端可以用 readEventName 取回，便于在同一队列里区分不同事件。
• ExcerptReader 实现了可关闭接口，用 try-with-resources 包住，避免资源泄漏。
• 队列使用完毕后调用 queue.close()，确保映射文件被正确释放。

第二步：写入结构化的多条消息

真实场景里一条消息通常包含多个字段。下面把行情 tick 抽象成 symbol、price、timestamp 三个字段，封装成一个生产者类，构造时获取一次 appender，之后反复复用。

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;

public class MarketDataProducer {
    private final ExcerptAppender appender;

    public MarketDataProducer(ChronicleQueue queue) {
        this.appender = queue.acquireAppender();
    }

    public void sendTick(String symbol, double price, long timestamp) {
        appender.writeDocument(w -> {
            w.writeEventName("tick")
             .writeString("symbol", symbol)
             .writeFloat("price", (float) price)
             .writeLong("timestamp", timestamp);
        });
    }
}

注意字段是按名字写入的，读取时必须用相同的名字和类型取回。价格这里用 writeFloat 写入，读取端就要用 readFloat，类型不匹配会导致解析错误。

第三步：读取并处理消息

读取端同样封装成一个类，构造时创建 reader，然后循环调用 readDocument。该方法返回布尔值，表示是否成功读到一条记录，因此可以放在 while 条件里持续消费。

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptReader;

public class MarketDataConsumer {
    private final ExcerptReader reader;

    public MarketDataConsumer(ChronicleQueue queue) {
        this.reader = queue.createReader();
    }

    public void processTicks() {
        while (reader.readDocument(r -> {
            String eventName = r.readEventName();
            String symbol = r.readString("symbol");
            float price = r.readFloat("price");
            long timestamp = r.readLong("timestamp");
            System.out.printf("Tick: %s @ %.2f%n", symbol, price);
        })) {
            // 继续读取下一条
        }
    }
}

读取顺序与写入顺序一致。如果队列里暂时没有新数据，readDocument 会返回 false，循环自然结束；需要持续跟随时，应改用 Tailer，见后文。

第四步：配置持久化与二进制编码

默认情况下队列已经落盘，但如果希望明确指定编码方式，可以在 builder 上设置 wireType。二进制编码更紧凑，适合对空间和解析速度有要求的场景。

ChronicleQueue queue = ChronicleQueue.builder()
        .path(Path.of("/data/persistent-queue"))
        .wireType(WireType.BINARY)
        .build();

消息写入后即持久化到磁盘，进程重启后队列内容仍然存在，恢复过程是自动的，不需要额外的恢复脚本。这一点对需要事后重放或审计的系统很关键。

第五步：设置滚动策略与文件大小上限

队列文件不会无限增长，而是按滚动周期切分。可以按天滚动，也可以设置单个文件的大小上限，两者可以同时使用。

ChronicleQueue queue = ChronicleQueue.builder()
        .path(Path.of("/data/roll-test"))
        .rollCycle(RollCycle.DAILY)          // 每天生成新文件
        .maxFileSize(256 * 1024 * 1024)      // 或文件达到 256MB 时滚动
        .build();

滚动周期决定了文件切分的粒度，也影响清理与归档的便利性。按天滚动便于按日期定位历史数据；按大小滚动则更适合写入量波动大的场景。具体可选的 RollCycle 取值以官网当前文档为准。

第六步：调整缓冲区容量

Chronicle Queue 默认性能已经不错，但在高吞吐写入时可以通过 bufferCapacity 调整缓冲区大小。

ChronicleQueue queue = ChronicleQueue.builder()
        .path(Path.of("/data/queue"))
        .bufferCapacity(1024)
        .build();

缓冲区容量需要结合消息大小与写入速率来定，盲目调大不一定带来收益，建议在真实负载下测量后再调整。

第七步：使用 Tailer 做流式消费

Reader 适合从头遍历已有数据，而 Tailer 的语义是「只读新消息」，更像一个流式消费者。构造方式与 reader 类似：

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptTailer;

public class StreamProcessor {
    private final ExcerptTailer tailer;

    public StreamProcessor(ChronicleQueue queue) {
        this.tailer = queue.createTailer();
    }

    public void process() {
        tailer.readDocument(r -> {
            String event = r.readEventName();
            handleEvent(event, r);
        });
    }
}

实际使用时通常把 readDocument 放进循环或调度中反复调用，以便持续跟进新写入的记录。

一个完整示例

下面把前面的片段串成一个可运行的最小系统：一个进程写入若干条 tick，另一个读取端从头读出并打印。为了便于演示，这里把写入与读取放在同一个 main 里，实际部署时应拆成独立进程。

import net.openhft.chronicle.queue.ChronicleQueue;
import net.openhft.chronicle.queue.ExcerptAppender;
import net.openhft.chronicle.queue.ExcerptReader;
import net.openhft.chronicle.queue.RollCycle;
import java.nio.file.Path;

public class TickDemo {
    public static void main(String[] args) {
        ChronicleQueue queue = ChronicleQueue.builder()
                .path(Path.of("/tmp/tick-demo"))
                .rollCycle(RollCycle.DAILY)
                .maxFileSize(256 * 1024 * 1024)
                .build();

        // 写入三条行情
        ExcerptAppender appender = queue.acquireAppender();
        writeTick(appender, "AAPL", 187.25, System.currentTimeMillis());
        writeTick(appender, "MSFT", 412.10, System.currentTimeMillis());
        writeTick(appender, "NVDA", 121.80, System.currentTimeMillis());

        // 从头读取并打印
        try (ExcerptReader reader = queue.createReader()) {
            while (reader.readDocument(r -> {
                String event = r.readEventName();
                String symbol = r.readString("symbol");
                float price = r.readFloat("price");
                long ts = r.readLong("timestamp");
                System.out.printf("%s %s @ %.2f (%d)%n", event, symbol, price, ts);
            })) {
                // 继续读取
            }
        }

        queue.close();
    }

    private static void writeTick(ExcerptAppender appender,
                                  String symbol, double price, long timestamp) {
        appender.writeDocument(w -> w
                .writeEventName("tick")
                .writeString("symbol", symbol)
                .writeFloat("price", (float) price)
                .writeLong("timestamp", timestamp));
    }
}

运行后，/tmp/tick-demo 目录下会出现队列文件。再次运行程序时，新的三条记录会追加到已有文件之后，读取端会读到全部历史记录——这正是持久化队列与内存队列的区别。若只想处理新数据，把读取端换成 Tailer 即可。

扩展：请求—响应模式

两个进程之间做请求—响应，可以用两个队列：一个放请求，一个放响应。请求端写入带 id 的消息，响应端读取后写回结果，请求端轮询响应队列直到超时。

public class RequestResponseQueue {
    private final ChronicleQueue requestQueue;
    private final ChronicleQueue responseQueue;

    public void request(String requestId, String payload) {
        requestQueue.acquireAppender().writeDocument(w -> w
                .writeString("id", requestId)
                .writeString("payload", payload));
    }

    public String waitForResponse(String requestId, long timeoutMs)
            throws InterruptedException, TimeoutException {
        long deadline = System.currentTimeMillis() + timeoutMs;
        try (ExcerptReader reader = responseQueue.createReader()) {
            while (System.currentTimeMillis() < deadline) {
                final String[] result = new String[1];
                if (reader.readDocument(r -> {
                    String id = r.readString("id");
                    if (id.equals(requestId)) {
                        result[0] = r.readString("response");
                    }
                }) && result[0] != null) {
                    return result[0];
                }
                Thread.sleep(1);
            }
        }
        throw new TimeoutException("No response within " + timeoutMs);
    }
}

这个模式的关键是给每条请求分配唯一 id，并在响应端原样带回，否则无法把响应与请求对应起来。轮询间隔与超时时间需要按业务延迟要求调整。

扩展：写入失败时回滚

写入过程中如果发现数据有问题，可以调用 appender 的回滚方法撤销当前这条记录，避免脏数据进入队列。

ExcerptAppender appender = queue.acquireAppender();
appender.writeDocument(w -> {
    w.writeEventName("trade")
     .writeLong("id", tradeId)
     .writeText("details", tradeDetails);
    // 校验失败时撤销本条记录
    // appender.rollback();
});

回滚应在写入闭包内部、根据校验结果决定是否调用，而不是无条件执行。

注意事项

适用边界

Chronicle Queue 的强项是单机内的低延迟与持久化，它不负责跨机器分发。以下情况适合使用它：

• 单机延迟比分布式特性更重要；
• 需要持久化，但不能接受网络或 broker 带来的额外延迟；
• 通信范围在同一个数据中心内，不涉及跨地域；
• 构建高频交易、交易平台或实时分析系统。

以下情况应考虑其他方案：

• 需要跨数据中心复制，可考虑 Kafka 或 Redpanda；
• 需要大规模水平扩展，可考虑 Kafka 或 Pulsar；
• 更倾向基于网络的通信而非共享内存，可考虑 Aeron。

常见坑

• 字段名与类型必须严格对应。 写入用 writeString("symbol", ...)，读取就必须用 readString("symbol")；名字拼错或类型不符都会导致读取失败。
• Reader 与 Tailer 语义不同。 Reader 会读到队列中已有的历史记录，Tailer 只关注新消息。用错会导致重复处理或漏处理。
• 队列需要显式关闭。 忘记 close() 可能造成文件映射未释放，影响后续进程打开同一目录。
• 滚动策略影响历史数据定位。 按天滚动后，跨天的数据分布在多个文件中，读取端需要能连续跨文件读取，不要假设只有一个文件。
• 磁盘空间需要监控。 持久化意味着数据会持续占用磁盘，滚动只改变文件切分方式，不会自动删除旧文件，清理策略需要自行设计。
• 版本与许可信息以官网为准。 依赖版本、企业版功能范围、价格与支持政策都可能变化，落地前请核对官网当前信息。

下一步可以做什么

把上面的示例跑通之后，可以尝试：修改滚动周期观察文件切分行为；用 Tailer 替换 Reader 验证流式消费；在真实负载下调整 bufferCapacity 并测量吞吐；以及把请求—响应模式拆成两个独立进程，验证跨进程通信。若需要复制等企业级能力，请查阅官网当前的功能说明。