使用ChronicleQueue创建低延迟的TB级别的队列D

　　本文介绍如何使用开源ChronicleQueue创建巨大的持久队列，同时保持可预测和一致的低延迟。
　　在本文中，目标是维护来自市场数据馈送的对象队列（例如，在交易所交易的证券的最新价格）。也可以选择其他业务领域，例如来自物联网设备的感官输入或读取汽车行业的碰撞记录信息。原理是一样的。
　　首先，定义一个持有市场数据的类：
　　publicclassMarketDataextendsSelfDescribingMarshallable｛intsecurityId；longtime；floatlast；floathigh；floatlow；Gettersandsettersnotshownforbrevity｝
　　注意：在实际场景中，使用float和double保存货币值时必须非常小心，否则可能会导致舍入问题〔Bloch18，Item60〕。但是，在这篇介绍性文章中，我想保持简单。
　　还有一个小的实用函数MarketDataUtil：：create将在调用时创建并返回一个新的随机MarketData对象：staticMarketDatacreate（）｛MarketDatamarketDatanewMarketData（）；intidThreadLocalRandom。current（）。nextInt（1000）；marketData。setSecurityId（id）；floatnextFloatThreadLocalRandom。current（）。nextFloat（）；floatlast20100nextFloat；marketData。setLast（last）；marketData。setHigh（last1。1f）；marketData。setLow（last0。9f）；marketData。setTime（System。currentTimeMillis（））；returnmarketData；｝
　　现在，目标是创建一个持久、并发、低延迟、可从多个进程访问并且可以容纳数十亿个对象的队列。
　　天真的方法
　　有了这些类，就可以探索使用ConcurrentLinkedQueue的简单方法：publicstaticvoidmain（String〔〕args）｛finalQueueMarketDataqueuenewConcurrentLinkedQueue（）；for（longi0；i1e9；i）｛queue。add（MarketDataUtil。create（））；｝｝
　　导致失败的有几个原因：ConcurrentLinkedQueue将为添加到队列中的每个元素创建一个包装节点。这将有效地使创建的对象数量增加一倍。对象放置在Java堆上，导致堆内存压力和垃圾收集问题。在我的机器上，这导致我的整个JVM变得无响应，唯一的办法是使用kill9强行杀死它。无法从其他进程（即其他JVM）读取队列。一旦JVM终止，队列的内容就会丢失。因此，队列不是持久的。
　　查看其他各种标准Java类，可以得出结论，不支持大型持久队列。
　　ChronicleQueue
　　ChronicleQueue是一个开源库，旨在满足上述要求。这是设置和使用它的一种方法：publicstaticvoidmain（String〔〕args）｛finalMarketDatamarketDatanewMarketData（）；finalChronicleQueueqChronicleQueue。single（marketdata）；finalExcerptAppenderappenderq。acquireAppender（）；for（longi0；i1e9；i）｛try（finalDocumentContextdocumentappender。acquireWritingDocument（false））｛document。wire（）。bytes（）。writeObject（MarketData。class，MarketDataUtil。recycle（marketData））；｝｝｝
　　使用配备2。3GHz8核英特尔酷睿i9的MacBookPro2019时，仅使用一个线程即可插入每秒超过3，000，000条消息。队列通过给定目录marketdata中的内存映射文件持久化。人们会期望MarketData对象至少占用4（intsecurityId）8（longtime）43（floatlast，high和low）24字节。
　　在上面的示例中，附加了10亿个对象，导致映射文件占用30，148，657，152个字节，这意味着每条消息大约30个字节。在我看来，这确实非常有效。
　　从编年史队列中读取很简单。继续上面的示例，下面显示了如何从队列中读取前两个MarketData对象：publicstaticvoidmain（String〔〕args）｛finalChronicleQueueqChronicleQueue。single（marketdata）；finalExcerptTailertailerq。createTailer（）；for（longi0；i2；i）｛try（finalDocumentContextdocumenttailer。readingDocument（））｛MarketDatamarketDatadocument。wire（）。bytes（）。readObject（MarketData。class）；System。out。println（marketData）；｝｝｝
　　还有许多其他功能超出了本文的范围。例如，可以将队列文件设置为以特定间隔（例如每天、每小时或每分钟）滚动，从而有效地创建信息分解，以便随着时间的推移清理旧数据。还有一些规定能够隔离CPU并将Java线程锁定到这些隔离的CPU，从而显着减少应用程序抖动。
　　ChronicleQueue更多：使用ChronicleQueue创建低延迟的TB级别的队列DZone