酷应用

Chronicle Queue入门

百家作者：51CTO技术栈 2022-10-07 08:29:19

作者丨Shaolang Ai

译者 | 杨晓娟

用Chronicle Queue构建的应用程序不会让生产者放慢将消息放入队列的速度（没有背压机制）。

Chronicle Queue(编年史队列)是低延迟、无代理、持久的消息队列。

与其最相近的是0MQ，但0MQ不存储发布的消息。Chronicle Queue的开源版本不支持跨机器通信。Chronicle Queue最与众不同之处在于它使用RandomAccessFile做堆外存储因而不会产生垃圾。

Chronicle Queue是以生产者为中心的，也就是说，用它构建的应用程序不会让生产者放慢将消息放入队列的速度（没有背压机制）。这种设计在对生产者的生产能力几乎不可控的情况下非常有用，例如外汇价格更新。

术语

大多数消息队列使用术语Producer（生产者）和Consumer（消费者），Chronicle Queue使用Appender（附加器）和Tailer（零售商），用于区分它总是将消息附加到队列中，并且零售商从队列中读取消息之后，从不“销毁/丢弃”任何消息。与Message(消息)相比，Chronicle Queue更喜欢使用术语Excerpt(摘录)，因为写入Chronicle Queue的blob可以是字节数组、字符串以及域模型。

Hello, World！

我们用传统的“Hello, World！”来演示基本用法。如果你使用的是Gradle，将以下内容添加到build.gradle.kts：

import?org.jetbrains.kotlin.gradle.tasks.KotlinCompile??//?line?1
?
plugins?{
???id("org.jetbrains.kotlin.jvm")?version?"1.3.71"
??application
}
?
repositories?{
??mavenCentral()
??mavenLocal()
}
?
dependencies?{
???implementation("org.jetbrains.kotlin:kotlin-bom")
???implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
implementation("net.openhft.chronicle:chronicle-queue:5.19.8")??//?line?17
?
implementation("org.apache.logging.log4j:log4j-sl4fj18-impl:2.13.1")
}
application?{
??mainClass?=??"hello.AppKt"
}
?
tasks.withType<KotlinCompile>?{??????????//?line?25
?????kotlinOptions.jvmTarget?=?"1.8"
}

左右滑动查看完整代码

导入KotlinCompile（第1行）允许将Java 1.8指定为编译目标（第25-27行）。第17-18行显示了开始使用Chronicle Queue所需的其它依赖项。请注意build.gradle.kts假定要使用的包是hello。接下来看看演示Chronicle Queue用法的代码：

package?hello
?
import?net.openhft.chronicle.queue.ChronicleQueue
?
fun?main(args:?Array<String>)?{
??val?q:??ChronicleQueue?=?ChronicleQueue.single("./build/hello-world")
?
??try?{
????val?appender:??ExcerptAppender?=?q.acquireAppender()
?????appender.writeText("Hello,?World!")
?????
????val?tailer:??ExcerptTailer?=?q.createTailer()
?????println(tailer.readText())
??}?finally?{
???q.close()
??}
}

左右滑动查看完整代码

ChronicleQueue.single(<path>)返回一个新建的使用给定的路径存储摘录的ChronicleQueue。其余的代码几乎是不言自明的：获得的appender把摘录“Hello, World!”追加到排队中；tailer从队列中读取并将摘录打印到标准输出。程序结束时一定要关闭队列。

还记得Chronicle Queue是持久的吗？注释掉两个appender行，然后再用gradle run执行程序。你将看到程序还是在标准输出上打印了Hello, World!：tailer读取的是上次运行时写入到队列中的数据。它的持久性允许在tailers崩溃时重放收到的摘录。

便道：摘录类型

Chronicle Queue仅接受以下类型的摘要：

Serializable对象：请注意，由于依赖于反射，序列化类对象的效率很低
Externalizable对象：如果与Java的兼容性很重要，但以牺牲手写逻辑为代价
net.openhft.chronicle.wire.Marshallable对象：使用二进制格式的高性能数据交换
net.openhft.chronicle.bytes.BytesMarshallable对象：底层二进制或文本编码

“Hello, World!”演示了字符串，我们顺便看一个使用Chronicle Wire库中Marshallable的例子。

package?types
?
import?net.openhft.chronicle.wire.Marshallable
import??net.openhft.chronicle.wire.SelfDescribingMarshallable
?
class?Person(val?name:?String,?val?age:?Int):??SelfDescribingMarshallable()
?
fun?main(args:?Array<String>)?{
????val?person?=?Person("Shaolang",?3)
????val?outputString?=?"""
????!types.Person?{
??????name:?Shaolang
??????age:?3
????}
????""".trimIndent()
?
????println(person.toString()?==?outputString)
?
????val?p?=?Marshallable.fromString<Person>(outputString)
????println(person?==?p)
????println(person.hashCode()?==?p.hashCode())
}

左右滑动查看完整代码

运行上面的代码片段会看到标准输出上打印了三个true。SelfDescribtingMarshallable可以轻松持久化Chronicle Queue中的Marshallable类。

写入和读取域对象

有了从上面小便道得来的经验，下面将演示向Chronicle Queue写入和读取Marshallable对象：

package?docs
?
import?net.openhft.chronicle.queue.ChronicleQueue
import??net.openhft.chronicle.wire.SelfDescribingMarshallable
?
class?Person(var?name:?String??=?null,?var?age:?Int??=??null):?SelfDescribingMarshallable()
?
class?Food(var?name:?String??=?null):??SelfDescribingMarshallable()
?
fun?main(args:?Array<String>)?{
????ChronicleQueue.single("./build/documents").use??{?q?->
????????val?appender?=??q.acquireAppender()?????????????????????????????????????????????????????
?????????appender.writeDocument(Person("Shaolang",?3))
????????appender.writeText("Hello,??World!")
????????appender.writeDocument(Food("Burger"))
?????????????????????????????????????????????????????
????????val?tailer?=?q.createTailer()
?????????????????????????????????????????????????????
????????val?person?=?Person()????????????????????????????????????????????????????
????????tailer.readDocument(person)
????????println(person)
?????????println("${tailer.readText()}\n")
?????????????????????????????????????????????????????
????????val?food?=?Food()
????????tailer.readDocument(food)
????????println(food)
????}
}

左右滑动查看完整代码

尽管在不同的VM进程中运行appender和tailer会更有意义，但将两者保持在同一个VM中可以更容易理解讨论，不必筛选无关的代码。运行上面的代码会看到如下输出：

!docs.Person?{
??name:?Shaolang,
??age:?3
}
Hello,?World!
!docs.Food?{
??name:?Burger,
}

有几点需要注意：

由于Chronicle Queue的目标是不产生垃圾，因而要求域模型是可变对象；这就是为什么两个类在构造器中使用var而不是val。
Chronicle ? ? Queue允许appender将不同的内容写入同一队列。
tailer需要知道它应该读什么才能得到正确的结果。

如果我们把最后一个tailer.readDocument(food)改成tailer.readDocument(person)然后打印person，将看到以下打印内容（至少在Chronicle Queue 5.19.x中，它不会崩溃/抛出任何异常）：

!docs.Person?{
??name:?Burger,
??age:?!!null?""
}

因为Person和Food有一个同名的属性，Chronicle Queue会尽可能匹配Person，不能匹配的置为空。

上面注意事项中的最后一点“关于tailer需要知道他们在读什么”会有点麻烦：它们(tailer)现在背负着过滤的重担，要从生产者不断扔来的雪崩一样的数据中获得它们想要的信息。为了保持代码库稳健，我们需要使用观察者模式。

只听感兴趣的东西

除了直接使用摘录附加器，另一种方法是使它具体化为传给methodWriter方法的第一类。下面的片段重点介绍指定侦听器的具体化：

package?listener
?
import?net.openhft.chronicle.queue.ChronicleQueue
import?net.openhft.chronicle.queue.ChronicleReaderMain
import??net.openhft.chronicle.wire.SelfDescribingMarshallable
?
class?Person(var?name:?String??=?null,?var?age:?Int??=??null):?SelfDescribingMarshallable()
?
interface?PersonListener?{
????fun?onPerson(person:?Person)
}
?
fun?main(args:?Array<String>)?{
????val?directory?=?"./build/listener"
?
?????ChronicleQueue.single(directory).use?{?q?->
????????val??observable:?PersonListener?=?q.acquireAppender()
??????????????.methodWriter(PersonListener::class.java)
?????????observable.onPerson(Person("Shaolang",?3))
?????????observable.onPerson(Person("Elliot",?4))
????}
?
??????ChronicleReaderMain.main(arrayOf("-d",?directory))
}

左右滑动查看完整代码

第17-18行用指定的PersonListener调用methodWriter获得附加器。请注意，赋予observable的类型是PersonListener，不是ExcerptAppender。现在，任何对PersonListener的方法调用都会把给定的参数写入队列。但是，直接使用附加器写入队列和使用具体化的类写入队列是有区别的。为了看出区别，我们使用ChronicleReaderMain检验队列：

0x47c900000000:
onPerson?{
??name:?Shaolang,
??age:?3
}
0x47c900000001:
onPerson?{
??name:?Elliot,
??age:?4
}

注意，具体化类写入队列的摘录用的是onPerson { ...} 而不是!listener.Person { ... }。这种差异允许实现了PersonListener的tailer收到写入队列的新Person对象的通知并忽略它们不感兴趣的对象。

是的，你没看错：实现了PersonListener的tailer。不幸的是，Chronicle Queue（有点）将被观察者和观察者混为一谈，因此很难区分它们。我认为区分差异的最简单方法是使用以下片段注释中所示的启发式方法：

interface?PersonListener?{
????onPerson(person:?Person)
}
//?this?is?an?observer?because?it?implements?the?listener??interface
class?PersonRegistry:?PersonListener?{
????override?fun?onPerson(person:?Person)?{
????????//?code?omitted?for?brevity
????}
}
fun?main(args:?Array<String>)?{
????//?code?omitted?for?brevity
????val??observable:?PersonListener?=?q.acquireAppender()????//?this?is?an
??????????????.methodWriter(PersonListener::class.java)???????//??observable
????//??another?way?to?differentiate:?the?observer?will?never?call?the
????//??listener?method,?only?observables?do.
?????observable.onPerson(Person("Shaolang",?3))
????//?code?omitted?for?brevity
}

左右滑动查看完整代码

再来看一下tailer。尽管Chronicle Queue确保每个tailer能看到每一条摘录，通过实现侦听器类/接口并用已实现的侦听器创建net.openhft.chronicle.bytes.MethodReader， tailer可以仅过滤出它想看到的摘录：

package?listener
?
import?net.openhft.chronicle.bytes.MethodReader
import?net.openhft.chronicle.queue.ChronicleQueue
import??net.openhft.chronicle.wire.SelfDescribingMarshallable
?
class?Person(var?name:?String??=?null,?var?age:?Int??=??null):?SelfDescribingMarshallable()
?
class?Food(var?name:?String??=?null):?SelfDescribingMarshallable()
?
interface?PersonListener?{
????fun?onPerson(person:?Person)
}
?
class?PersonRegistry:?PersonListener?{
????override?fun?onPerson(person:?Person)?{
????????println("in?registry:??${person.name}")
????}
}
?
fun?main(args:?Array<String>)?{
?????ChronicleQueue.single("./build/listener2").use?{?q?->
????????val?appender?=??q.acquireAppender()
????????val?writer:?PersonListener?=??appender.methodWriter(PersonListener::class.java)
?????????writer.onPerson(Person("Shaolang",?3))
?????????appender.writeDocument(Food("Burger"))
?????????writer.onPerson(Person("Elliot",?4))
?????????????????????????????????????????????????????
????????val?registry:?PersonRegistry?=??PersonRegistry()
????????val?reader:??MethodReader?=?q.createTailer().methodReader(registry)
????????reader.readOne()
????????reader.readOne()
????????reader.readOne()
????}
}

左右滑动查看完整代码

这里的主要新内容是PersonRegistry的实现，它简单地打印出所给的person的name。代码片段并没直接用ExcerptTailer从队列中读取而是用给定的PersonRegistry由tailer创建了一个MethodReader。?

.methodWriter接受Class参数,而.methodReader接受的是对象。appender向队列写入三个摘录：person（通过调用onPerson)、food（通过.writeDocument)和person。因为tailer可以看到每一个摘录，所以阅读者也会调用三次“读取”所有摘录，但却只会看到两个输出：

in registry：Shaolang

in registry：Elliot

如果代码片段只有两个.readOne()调用而不是三个，那么输出中就不会包含in registry：Elliot.

MethodReader使用鸭子类型

还记得我们检验由具体化的PersonListener填充队列时ChronicleReaderMain的输出吗？输出的不是类名而是类似于onPerson { ... }。这表明MethodReader过滤与方法签名匹配的摘录，即它不关心包含方法签名的接口/类；或者简单地说，鸭子类型:

package?duck
?
import?net.openhft.chronicle.queue.ChronicleQueue
import??net.openhft.chronicle.wire.SelfDescribingMarshallable
?
class?Person(var?name:?String??=?null,?var?age:?Int??=??null):?SelfDescribingMarshallabl()
?
interface?PersonListener?{
????fun?onPerson(person:?Person)
}
?
interface?VIPListener?{
????fun?onPerson(person:?Person)
}
?
class?VIPClub:?VIPListener?{
????override?fun??onPerson(person:?Person)?{
????????println("Welcome?to?the??club,?${person.name}!")
????}
}
?
fun?main(args:?Array<String>)?{
????ChronicleQueue.single("./build/duck").use??{?q?->
????????val?writer?=??q.acquireAppender().methodWriter(PersonListener::class.java)
?????????writer.onPerson(Person("Shaolang",?3))
????????????????????????????????????????????????
????????val?club?=?VIPClub()
????????val?reader?=??q.createTailer().methodReader(club)
????????reader.readOne()
????}
}

左右滑动查看完整代码

注意，VIPClub实现了VIPListener，碰巧与PersonListener有相同的onPerson方法签名。运行上面的代码，你会看到打印的Welcome to the club, Shaolang!

命名tailer

到目前为止，在所有的演示中，我们一直创建的都是匿名的tailer。因为它们是匿名的，所以每次（重新）运行都会读取队列中的所有摘录。有时，这样的行为是可接受的，甚至是可取的，但有时却不是。只需命名tailer就可以从上次停止的位置继续读取：

package?restartable
?
import?net.openhft.chronicle.queue.ChronicleQueue
import?net.openhft.chronicle.queue.ExcerptTailer
?
fun?readQueue(tailerName:?String,?times:?Int)?{
?????ChronicleQueue.single("./build/restartable").use?{?q?->
?????????val?tailer?=?q.createTailer(tailerName)???????//?tailer??name?given
?????????for?(_n?in?1..times)?{
???????????println("$tailerName:?${tailer.readText()}")
????????}
?
????????println()????????//?to?separate?outputs?for?easier?visualization
????}
}
?
fun?main(args:?Array<String>)?{
?????ChronicleQueue.single("./build/restartable").use?{?q?->
????????val?appender?=??q.acquireAppender()
????????appender.writeText("Test??Message?1")
????????appender.writeText("Test??Message?2")
????????appender.writeText("Test?Message??3")
????????appender.writeText("Test??Message?4")
}
?
????readQueue("foo",?1)
????readQueue("bar",?2)
????readQueue("foo",?3)
????readQueue("bar",?1)
}

左右滑动查看完整代码

注意，tailer的名字是通过createTailer方法指定的。上面的代码中有两个tailer（命名为foo和bar）读取队列并在运行时输出以下内容：

foo:?Test?Message?1
bar:?Test?Message?1
bar:?Test?Message?2
foo:?Test?Message?2
foo:?Test?Message?3
foo:?Test?Message?4
bar:?Test?Message?3?

注意，foo和bar第二次从队列中读取数据时，会从之前断开的位置开始。

滚动文件

Chronicle Queue根据创建队列时定义的滚动周期滚动使用的文件；默认情况下，每天滚动文件。要改变滚动周期，就不能使用简单的ChronicleQueue.single方法：

package?roll
?
import?net.openhft.chronicle.queue.ChronicleQueue
import?net.openhft.chronicle.queue.RollCycles
import?net.openhft.chronicle.impl.single.SingleChronicleQueueBuilder
?
fun?main(args:?Array<String>)?{
????var??qbuilder:?SingleChronicleQueueBuilder?=??ChronicleQueue.singleBuilder("./build/roll")
?????qbuilder.rollCycle(RollCycles.HOURLY)
????val?q:??ChronicleQueue?=?qbuilder.build()
????//?code??omitted?for?brevity
}

左右滑动查看完整代码

首先，得到一个SingleChronicleQueueBuilder实例，并通过.rollCycle方法设置滚动周期。上面的代码段将队列配置为每小时滚动一次文件。配置好后，调用构造器的.build（）获取ChronicleQueue实例。请注意，appender和tailer(s)在访问同一个队列时必须使用相同的滚动周期。

由于SingleChronicleQueueBuilder支持流式接口，代码也可以做如下简化：

val?q:?ChronicleQueue?=?ChronicleQueue.singleBuilder("./build/roll")
??????????????????????????????????????.rollCycle(RollCycles.HOURLY)
??????????????????????????????????????.build()

左右滑动查看完整代码