作者丨Shaolang Ai
譯者 | 楊曉娟
用Chronicle Queue構(gòu)建的應用程序不會讓生產(chǎn)者放慢將消息放入隊列的速度（沒有背壓機制）。
Chronicle Queue(編年史隊列)是低延遲、無代理、持久的消息隊列。
與其最相近的是0MQ，但0MQ不存儲發(fā)布的消息。Chronicle Queue的開源版本不支持跨機器通信。Chronicle Queue最與眾不同之處在于它使用RandomAccessFile做堆外存儲因而不會產(chǎn)生垃圾。
Chronicle Queue是以生產(chǎn)者為中心的，也就是說，用它構(gòu)建的應用程序不會讓生產(chǎn)者放慢將消息放入隊列的速度（沒有背壓機制）。這種設計在對生產(chǎn)者的生產(chǎn)能力幾乎不可控的情況下非常有用，例如外匯價格更新。
術(shù)語
大多數(shù)消息隊列使用術(shù)語Producer（生產(chǎn)者）和Consumer（消費者），Chronicle Queue使用Appender（附加器）和Tailer（零售商），用于區(qū)分它總是將消息附加到隊列中，并且零售商從隊列中讀取消息之后，從不“銷毀/丟棄”任何消息。與Message(消息)相比，Chronicle Queue更喜歡使用術(shù)語Excerpt(摘錄)，因為寫入Chronicle Queue的blob可以是字節(jié)數(shù)組、字符串以及域模型。
Hello, World！
我們用傳統(tǒng)的“Hello, World！”來演示基本用法。如果您使用的是Gradle，將以下內(nèi)容添加到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()返回一個新建的使用給定的路徑存儲摘錄的ChronicleQueue。其余的代碼幾乎是不言自明的：獲得的appender把摘錄“Hello, World!”追加到排隊中；tailer從隊列中讀取并將摘錄打印到標準輸出。程序結(jié)束時一定要關(guān)閉隊列。
還記得Chronicle Queue是持久的嗎？注釋掉兩個appender行，然后再用gradle run執(zhí)行程序。您將看到程序還是在標準輸出上打印了Hello, World!：tailer讀取的是上次運行時寫入到隊列中的數(shù)據(jù)。它的持久性允許在tailers崩潰時重放收到的摘錄。
便道：摘錄類型
Chronicle Queue僅接受以下類型的摘要：
1. Serializable對象：請注意，由于依賴于反射，序列化類對象的效率很低
2. Externalizable對象：如果與Java的兼容性很重要，但以犧牲手寫邏輯為代價
3. net.openhft.chronicle.wire.Marshallable對象：使用二進制格式的高性能數(shù)據(jù)交換
4. 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類。
寫入和讀取域?qū)ο?/h2>
有了從上面小便道得來的經(jīng)驗，下面將演示向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中可以更容易理解討論，不必篩選無關(guān)的代碼。運行上面的代碼會看到如下輸出：
!docs.Person {
name: Shaolang,
age: 3
}
Hello, World!
!docs.Food {
name: Burger,
}有幾點需要注意：
1. 由于Chronicle Queue的目標是不產(chǎn)生垃圾，因而要求域模型是可變對象；這就是為什么兩個類在構(gòu)造器中使用var而不是val。
2. Chronicle Queue允許appender將不同的內(nèi)容寫入同一隊列。
3. tailer需要知道它應該讀什么才能得到正確的結(jié)果。
如果我們把最后一個tailer.readDocument(food)改成tailer.readDocument(person)然后打印person，將看到以下打印內(nèi)容（至少在Chronicle Queue 5.19.x中，它不會崩潰/拋出任何異常）：
!docs.Person {
name: Burger,
age: !!null ""
}因為Person和Food有一個同名的屬性，Chronicle Queue會盡可能匹配Person，不能匹配的置為空。
上面注意事項中的最后一點“關(guān)于tailer需要知道他們在讀什么”會有點麻煩：它們(tailer)現(xiàn)在背負著過濾的重擔，要從生產(chǎn)者不斷扔來的雪崩一樣的數(shù)據(jù)中獲得它們想要的信息。為了保持代碼庫穩(wěn)健，我們需要使用觀察者模式.
（有點）只聽感興趣的東西
除了直接使用摘錄附加器，另一種方法是使它具體化為傳給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調(diào)用methodWriter獲得附加器。請注意，賦予observable的類型是PersonListener，不是ExcerptAppender。現(xiàn)在，任何對PersonListener的方法調(diào)用都會把給定的參數(shù)寫入隊列。但是，直接使用附加器寫入隊列和使用具體化的類寫入隊列是有區(qū)別的。為了看出區(qū)別，我們使用ChronicleReaderMain檢驗隊列：
0x47c900000000:
onPerson {
name: Shaolang,
age: 3
}
0x47c900000001:
onPerson {
name: Elliot,
age: 4
}
注意，具體化類寫入隊列的摘錄用的是onPerson { ...} 而不是!listener.Person { ... }。 這種差異允許實現(xiàn)了PersonListener的tailer收到寫入隊列的新Person對象的通知并忽略它們不感興趣的對象。
是的，你沒看錯：實現(xiàn)了PersonListener的tailer。不幸的是，Chronicle Queue（有點）將被觀察者和觀察者混為一談，因此很難區(qū)分它們。我認為區(qū)分差異的最簡單方法是使用以下片段注釋中所示的啟發(fā)式方法：
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能看到每一條摘錄，通過實現(xiàn)偵聽器類/接口并用已實現(xiàn)的偵聽器創(chuàng)建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()
}
}
這里的主要新內(nèi)容是PersonRegistry的實現(xiàn)，它簡單地打印出所給的person的name。 代碼片段并沒直接用ExcerptTailer從隊列中讀取而是用給定的PersonRegistry由tailer創(chuàng)建了一個MethodReader。
.methodWriter接受Class參數(shù),而.methodReader接受的是對象。appender向隊列寫入三個摘錄：person（通過調(diào)用onPerson)、food（通過.writeDocument)和person。 因為tailer可以看到每一個摘錄，所以閱讀者也會調(diào)用三次“讀取”所有摘錄，但卻只會看到兩個輸出：
in registry：Shaolang
in registry：Elliot
如果代碼片段只有兩個.readOne()調(diào)用而不是三個，那么輸出中就不會包含in registry：Elliot.
MethodReader使用鴨子類型
還記得我們檢驗由具體化的PersonListener填充隊列時ChronicleReaderMain的輸出嗎？輸出的不是類名而是類似于onPerson { ... }。這表明MethodReader過濾與方法簽名匹配的摘錄，即它不關(guān)心包含方法簽名的接口/類；或者簡單地說，鴨子類型:
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實現(xiàn)了VIPListener，碰巧與PersonListener有相同的onPerson方法簽名。運行上面的代碼，你會看到打印的Welcome to the club, Shaolang!
命名tailer
到目前為止，在所有的演示中，我們一直創(chuàng)建的都是匿名的tailer。因為它們是匿名的，所以每次（重新）運行都會讀取隊列中的所有摘錄。有時，這樣的行為是可接受的，甚至是可取的，但有時卻不是。只需命名tailer就可以從上次停止的位置繼續(xù)讀取：
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）讀取隊列并在運行時輸出以下內(nèi)容：
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第二次從隊列中讀取數(shù)據(jù)時，會從之前斷開的位置開始。
滾動文件
Chronicle Queue根據(jù)創(chuàng)建隊列時定義的滾動周期滾動使用的文件；默認情況下，每天滾動文件。要改變滾動周期，就不能使用簡單的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方法設置滾動周期。 上面的代碼段將隊列配置為每小時滾動一次文件。配置好后，調(diào)用構(gòu)造器的.build（）獲取ChronicleQueue實例。請注意，appender和tailer(s)在訪問同一個隊列時必須使用相同的滾動周期。
由于SingleChronicleQueueBuilder支持流式接口，代碼也可以做如下簡化：
val q: ChronicleQueue = ChronicleQueue.singleBuilder("./build/roll")
.rollCycle(RollCycles.HOURLY)
.build()
接下來
這篇文章介紹了Chronicle Queue的術(shù)語和基礎知識。以下網(wǎng)站有更多信息可供挖掘：
1. Chronicle Queue GitHub repository
2. Stack Overflow tagged questions
3. Peter Lawre's Blog
原文鏈接
https://dzone.com/articles/bit-by-bit
譯者介紹
楊曉娟，51CTO社區(qū)編輯，資深研發(fā)工程師，信息系統(tǒng)項目管理師，擁有近20年Java開發(fā)經(jīng)驗。
