队列通常是软件设计模式中的基本组件。但是如果每秒接收到数百万条消息，改如何处理？如果多个消费者都需要能够读取所有消息，又改如何处理？难道需要把所有消息的数据都放在内存中吗？这样 JVM GC 又表现如何？
之前我写过几个流量回放模型：
- 基于时间戳的日志回放引擎
2022-08-22 - 千万级日志回放引擎设计稿
2021-12-30
虽然方案 2 已经被更优秀的方案替代，但是思路相同，均是把日志进行格式转换之后存放（这一点跟 goreplay 略有相似），在千万日志级别，我是直接放在内存中。大约 1 千万日志的大小约为 1G，这样来说对 JVM 内存压力并不高，对于 GC 的影响也可以接受，目前的测试结果是 YoungGC 1次/3s，全程无 FullGC。
但是如果想要更近一步，实现更大规模的日志回放，就不能采取这种方式，需要把日志存在磁盘中，用的时候顺序读取，这个速度大概 80 万/s。也算是满足需求了。但是其中需要使用java.lang.String#split(java.lang.String, int)，又比较消耗性能。
这个时候接触了Chronicle Queue，看了简介，简直爆炸，而且 API 简单好用，性能又高。特别是支持 TB 级别文件高性能、低延迟的读写。太符合我的需求了。后续我再根据实际情况进行实践、测试、分享。
本文介绍如何使用 Chronicle Queue 创建巨大的持久队列，同时保持可预测和一致的低延迟。
演示
在本文中，我维护一个保留日志回放的日志队列，首先是一个日志类，对原来的文章进行了一些Chronicle Queue化改造，保留了日志时间戳、host等信息。
代码语言：javascript复制 private static class FunLog extends SelfDescribingMarshallable {
String url
String host
int time
FunLog() {
}
FunLog(String url, String host, int time) {
this.url = url
this.host = host
this.time = time
}
}
官方提醒：字段值为浮点类型时，切记注意有效位数长度问题。有兴趣的可以看一看Java 序列化10倍性能优化对比测试关于Chronicle Queue序列化相关方案。
最初的方案
首先想到了探索使用 ConcurrentLinkedQueue 的方法：
代码语言：javascript复制public static void main(String[] args) {
final Queue<MarketData> queue = new ConcurrentLinkedQueue<>();
for (long i = 0; i < 1e9; i ) {
queue.add(new FunLog(Time.getDate(), index.getAndIncrement() EMPTY, getMark()));
}
}
但是最终将会崩溃，有几个原因：
- ConcurrentLinkedQueue 将为添加到队列中的每个元素创建一个包装节点。这将使创建的对象数量增加一倍。
- 对象放置在 Java 堆上，导致堆内存压力和垃圾收集问题，很可能导致卡死，只能强制结束进程。
- 无法从其他进程（即其他 JVM）读取队列。
- 一旦 JVM 终止，队列的内容就会丢失，队列不是持久化的。
其他各种标准 Java 类，均是不支持大型持久队列。
Chronicle Queue
Chronicle Queue 是一个开源库，旨在满足上述要求。这是设置和使用它的一种方法：
代码语言：javascript复制 static void main(String[] args) {
String basePath = getLongFile("chronicle")
ChronicleQueue queue = ChronicleQueue.singleBuilder(basePath).build()
def appender = queue.acquireAppender()
int total = 1_0000_0000
def start = Time.getTimeStamp()
total.times {
def log = new FunLog(Time.getDate(), index.getAndIncrement() EMPTY, getMark())
appender.writeDocument(log)
}
def end = Time.getTimeStamp()
output(total / (end - start) * 1000)
output(queue.lastIndex() - queue.firstIndex())
}
由于不可描述的原因，我本机的 IO 性能被降低了很多，但是在使用以上用例创建一个长度 1 亿的队列时，Chronicle Queue还是表现了非常好的性能，平均的 QPS 为 170 万，占用磁盘空间 4.5G，而且读取速度也保持在 160 万 QPS 量级。
读取用例如下：
代码语言：javascript复制 static void main(String[] args) {
String basePath = getLongFile("chronicle")
ChronicleQueue queue = ChronicleQueue.singleBuilder(basePath).build()
def tailer = queue.createTailer()
def log = new FunLog()
int total = 1_0000_0000
def start = Time.getTimeStamp()
total.times {
tailer.readDocument(log)
}
def end = Time.getTimeStamp()
output(total / (end - start) * 1000)
output(queue.lastIndex() - queue.firstIndex())
}
可以看出，我只用了一个com.funtest.queue.Qt.FunLog对象，这样就进一步降低了 JVM 内存和 GC 的压力。当然我们写入队列时，也可以使用这样的方式，不过在我的设计中，直接读取日志文件进行格式转换，可以直接使用通用池化框架GenericObjectPool性能测试、通用池化框架GenericKeyedObjectPool性能测试，后面有时间再来分享。
下面是我两次测试的 JVM 监控截图，可见Chronicle Queue的强大：
