省编码市编码区县编码

本文介绍了如何通过减少软件堆栈中的浪费来高效解决有意义的事件处理问题。

Java通常被视为无法在低内存环境中有效运行的内存猪。目的是证明许多人认为不可能的事情，有意义的java程序几乎可以在没有内存的情况下运行。示例流程
在Java的单个线程上，零内存gc的3MB堆中每秒有220万个csv记录 。

您将了解Java应用程序中主要浪费区域的位置以及可以用来减少浪费的模式。引入了零成本抽象的概念，并且可以通过代码生成在编译时自动进行许多优化。 Maven插件简化了开发人员的工作流程。

我们的目标不是高性能，而是高效率的副产品。该解决方案采用了Fluxtion ，与现有的Java事件处理框架相比，它占用的资源很少。

计算与气候

当前，气候变化及其原因引起许多人的极大关注。计算是主要的排放源，产生的碳足迹与整个航空业相同。在没有规定计算能耗的法规的情况下，作为工程师，我们必须承担生产与成本相平衡的高效系统的责任。

在伦敦Infoq 2019的一次小组会议上，马丁·汤普森（ Martin Thompson）热情洋溢地谈到了建筑节能计算系统。他指出，控制废物是最大程度降低能耗的关键因素。 Martin的评论引起了我的共鸣，因为Fluxtion背后的核心理念是消除不必要的资源消耗。小组会议是本文的灵感。

加工要求

处理示例的要求是：

使用零gc在3MB的堆中运行
仅使用标准Java库，没有“不安全”的优化
读取包含数百万行输入数据的CSV文件
输入是一组未知事件，没有预先加载数据
数据行是异构类型
处理每一行以计算多个汇总值
计算取决于行类型和数据内容
将规则应用于汇总并计算违反规则的次数
数据随机分配以防止分支预测
根据行输入值进行分区计算
收集分区计算并将其分组到汇总视图中
在文件末尾发布摘要报告
使用高级功能的纯Java解决方案
没有准时热身

头寸和利润监控示例

CSV文件包含一系列资产的交易和价格，每行一条记录。每个资产的头寸和利润计算都划分在其自己的内存空间中。资产计算会在每个匹配的输入事件上更新。所有资产的利润将汇总为投资组合利润。每项资产都监视其当前头寸/利润状态，并在其中一项违反预设限制时记录计数。投资组合的利润将受到监控，并计算违约损失。

针对每个传入事件在资产和投资组合级别验证规则。随着事件流式传输到系统中，违反规则的计数也会更新。

行数据类型

href="https://github.com/gregv12/articles/blob/article_may2019/2019/may/trading-monitor/src/main/java/com/fluxtion/examples/tradingmonitor/AssetPrice.java" target="_blank" rel="noopener noreferrer">AssetPrice - [price: double] [symbol: CharSequence]

 Deal       - [price: double] [symbol: CharSequence] [size: int]

样本数据

CSV文件的每种类型都有一个标题行，以允许动态列位置映射到字段。每行前面都有要编组的目标类型的简单类名。记录示例集，包括标题：

Deal,symbol,size,price
AssetPrice,symbol,price
AssetPrice,FORD,15.0284
AssetPrice,APPL,16.4255
Deal,AMZN,-2000,15.9354

计算说明

资产计算按符号划分，然后收集到资产组合计算中。

分区资产计算

asset position  = sum(Deal::size)
deal cash value = (Deal::price) X (Deal::size) X -1
cash position   = sum(deal cash value)
mark to market  = (asset position) X (AssetPrice::price)
profit          = (asset mark to market) + (cash position)

投资组合计算

portfolio profit = sum(asset profit)

监控规则

asset loss > 2,000
asset position outside of range +- 200
portfolio loss > 10,000

注意：

当通知程序指示违反规则时，将进行计数。通知程序仅在第一个违规时触发，直到将其重置。当规则再次变得有效时，将重置通知程序。
正交易::大小是买入，负值是卖出。

执行环境

为了确保满足内存要求（零gc和3MB堆），
使用Epsilon无操作垃圾收集器，最大堆大小为3MB。如果在整个过程的生命周期中分配了超过3MB的内存，则JVM将立即退出，并显示内存不足错误。

运行示例：从git克隆，并在trading-monitor项目的根目录中，运行dist目录中的jar文件，以生成400万行的测试数据文件。

git clone --branch  article_may2019 https://github.com/gregv12/articles.git
cd articles/2019/may/trading-monitor/
jdk-12.0.1\bin\java.exe -jar dist\tradingmonitor.jar 4000000

默认情况下，tradingmonitor.jar处理data / generated-data.csv文件。使用上面的命令，输入数据应具有400万行，并且长度为94MB，可以执行。

结果

要执行测试，请运行不带参数的tradingmonitor.jar：

jdk-12.0.1\bin\java.exe -verbose:gc -Xmx3M -XX:+UnlockExperimentalVMOptions -XX:+UseEpsilonGC -jar dist\tradingmonitor.jar

对400万行执行测试，摘要结果为：

Process row count     =    4 million
Processing time       =    1.815 seconds
Avg row exec time     =  453 nano seconds
Process rate          =    2.205 million records per second
garbage collections   =    0
allocated mem total   = 2857 KB
allocated mem per run =   90 KB
OS                    = windows 10
Processor             = Inte core i7-7700@3.6Ghz
Memory                = 16 GB
Disk                  = 512GB Samsung SSD PM961 NVMe

注意：结果来自没有JIT预热的第一次运行。 jit预热后，代码执行时间缩短了大约10％。分配的总内存为2.86Mb，其中包括启动JVM。

通过分析Epsilon的输出，我们估计应用程序为6次运行分配了15％的内存，即每次运行分配90KB。应用程序数据很可能适合L1缓存，此处需要进行更多调查。

输出量

每次测试程序循环打印6次结果，Epsilon在运行结束时记录内存统计信息。

jdk-12.0.1\bin\java.exe" -server -XX:+UnlockExperimentalVMOptions -XX:+UseEpsilonGC  -Xmx3M -verbose:gc -jar dist\tradingmonitor.jar
[0.011s][info][gc] Non-resizeable heap; start/max: 3M
[0.011s][info][gc] Using TLAB allocation; max: 4096K
[0.011s][info][gc] Elastic TLABs enabled; elasticity: 1.10x
[0.011s][info][gc] Elastic TLABs decay enabled; decay time: 1000ms
[0.011s][info][gc] Using Epsilon
[0.024s][info][gc] Heap: 3M reserved, 3M (100.00%) committed, 0M (5.11%) used
[0.029s][info][gc] Heap: 3M reserved, 3M (100.00%) committed, 0M (10.43%) used
.....
.....
[0.093s][info][gc] Heap: 3M reserved, 3M (100.00%) committed, 1M (64.62%) used
[0.097s][info][gc] Heap: 3M reserved, 3M (100.00%) committed, 2M (71.07%) usedportfolio loss gt 10k count -> 792211.0
Portfolio PnL:-917.6476000005273
Deals processed:400346
Prices processed:3599654
Assett positions:
-----------------------------
[1.849s][info][gc] Heap: 3M reserved, 3M (100.00%) committed, 2M (76.22%) used
MSFT : AssetTradePos{symbol=MSFT, pnl=484.68589999993696, assetPos=97.0, mtm=1697.0247000000002, cashPos=-1212.3388000000632, positionBreaches=139, pnlBreaches=13628, dealsProcessed=57046, pricesProcessed=514418}
GOOG : AssetTradePos{symbol=GOOG, pnl=-998.6065999999155, assetPos=-1123.0, mtm=-19610.1629, cashPos=18611.556300000084, positionBreaches=3, pnlBreaches=105711, dealsProcessed=57199, pricesProcessed=514144}
APPL : AssetTradePos{symbol=APPL, pnl=-21.881300000023202, assetPos=203.0, mtm=3405.1017, cashPos=-3426.9830000000234, positionBreaches=169, pnlBreaches=26249, dealsProcessed=57248, pricesProcessed=514183}
ORCL : AssetTradePos{symbol=ORCL, pnl=-421.9756999999504, assetPos=-252.0, mtm=-4400.4996, cashPos=3978.5239000000497, positionBreaches=103, pnlBreaches=97777, dealsProcessed=57120, pricesProcessed=513517}
FORD : AssetTradePos{symbol=FORD, pnl=112.14559999996254, assetPos=-511.0, mtm=-7797.8089, cashPos=7909.9544999999625, positionBreaches=210, pnlBreaches=88851, dealsProcessed=57177, pricesProcessed=514756}
BTMN : AssetTradePos{symbol=BTMN, pnl=943.8932999996614, assetPos=-1267.0, mtm=-19568.9417, cashPos=20512.83499999966, positionBreaches=33, pnlBreaches=117661, dealsProcessed=57071, pricesProcessed=514291}
AMZN : AssetTradePos{symbol=AMZN, pnl=-557.0849999999355, assetPos=658.0, mtm=10142.214600000001, cashPos=-10699.299599999937, positionBreaches=63, pnlBreaches=114618, dealsProcessed=57485, pricesProcessed=514345}
-----------------------------
Events proecssed:4000000
millis:1814
...
...
portfolio loss gt 10k count -> 792211.0
Portfolio PnL:-917.6476000005273
Deals processed:400346
Prices processed:3599654
Assett positions:
-----------------------------
MSFT : AssetTradePos{symbol=MSFT, pnl=484.68589999993696, assetPos=97.0, mtm=1697.0247000000002, cashPos=-1212.3388000000632, positionBreaches=139, pnlBreaches=13628, dealsProcessed=57046, pricesProcessed=514418}
GOOG : AssetTradePos{symbol=GOOG, pnl=-998.6065999999155, assetPos=-1123.0, mtm=-19610.1629, cashPos=18611.556300000084, positionBreaches=3, pnlBreaches=105711, dealsProcessed=57199, pricesProcessed=514144}
APPL : AssetTradePos{symbol=APPL, pnl=-21.881300000023202, assetPos=203.0, mtm=3405.1017, cashPos=-3426.9830000000234, positionBreaches=169, pnlBreaches=26249, dealsProcessed=57248, pricesProcessed=514183}
ORCL : AssetTradePos{symbol=ORCL, pnl=-421.9756999999504, assetPos=-252.0, mtm=-4400.4996, cashPos=3978.5239000000497, positionBreaches=103, pnlBreaches=97777, dealsProcessed=57120, pricesProcessed=513517}
FORD : AssetTradePos{symbol=FORD, pnl=112.14559999996254, assetPos=-511.0, mtm=-7797.8089, cashPos=7909.9544999999625, positionBreaches=210, pnlBreaches=88851, dealsProcessed=57177, pricesProcessed=514756}
BTMN : AssetTradePos{symbol=BTMN, pnl=943.8932999996614, assetPos=-1267.0, mtm=-19568.9417, cashPos=20512.83499999966, positionBreaches=33, pnlBreaches=117661, dealsProcessed=57071, pricesProcessed=514291}
AMZN : AssetTradePos{symbol=AMZN, pnl=-557.0849999999355, assetPos=658.0, mtm=10142.214600000001, cashPos=-10699.299599999937, positionBreaches=63, pnlBreaches=114618, dealsProcessed=57485, pricesProcessed=514345}
-----------------------------
Events proecssed:4000000
millis:1513
[14.870s][info][gc] Total allocated: 2830 KB
[14.871s][info][gc] Average allocation rate: 19030 KB/sec

废物热点

下表标识了处理循环中的功能，这些功能通常会创建示例中使用的浪费和避免浪费技术。

功能	废物来源	影响	回避
读取CSV文件	为每行分配一个新的字符串	GC	将每个字节读入一个flyweight，并在无分配解码器中进行处理
行数据持有人	为每一行分配一个数据实例	GC	Flyweight单个数据实例
读取列值	为每列分配一个字符串数组	GC	将字符推送到可重复使用的字符缓冲区中
将值转换为类型	字符串到类型的转换分配内存	GC	零分配转换器CharSequence代替字符串
将col值推送给持有人	基本类型的自动装箱会分配内存。	GC	原始感知功能可推送数据。零分配
分区数据处理	数据分区并行处理。分配给队列的任务	GC /锁	单线程处理，无分配或锁
计算方式	自动装箱，分配中间实例的不可变类型。无状态功能需要外部状态存储和分配	GC	生成没有自动装箱的功能。有状态功能零分配
汇总摘要计算	将分区线程的结果推送到队列中。需要分配和同步	GC /锁	单线程处理，无分配或锁

减少废物的解决方案

使用Fluxtion生成实现事件处理的代码。生成解决方案允许采用零成本抽象方法，其中已编译的解决方案的开销最少。程序员描述所需的行为，并在构建时生成满足要求的优化解决方案。对于此示例，可以在此处查看生成的代码。

maven pom包含一个配置文件，用于使用通过以下命令执行的Fluxtion maven插件重建生成的文件：

mvn -Pfluxtion install

文件读取

从输入文件中提取数据作为一系列CharEvents ，并将其发布到csv类型的marshaller。每个字符都可以从文件中单独读取，然后推入CharEvent中。由于重复使用了相同的CharEvent实例，因此初始化后不会分配任何内存。用于流CharEvents逻辑位于CharStreamer类。整个96 MB的文件可以读取，应用程序在堆上分配的内存几乎为零。

CSV处理

在Javabean中添加@CsvMarshaller会通知Fluxtion在构建时生成csv解析器。 Fluxtion在应用程序类中扫描@CsvMarshaller批注，并在构建过程中生成编组器。有关示例，请参见AssetPrice.java ，它会生成AssetPriceCsvDecoder0 。解码器处理CharEvents并将行数据编组到目标实例中。

生成的CSV解析器采用上表中概述的策略，避免了不必要的内存分配，并为处理的每一行重用了对象实例：

单个可重复使用的字符缓冲区实例存储行字符
轻量级可重用实例是编组列数据的目标
直接从CharSequence转换为目标类型，而无需创建中间对象。
如果在目标实例中使用了CharSequence，则不会创建任何字符串，则将使用一个轻量级的Charsequence。

有关将无用字符转换为目标字段的示例，请参见AssetPriceCsvDecoder中的upateTarget（）方法：

计算方式

该构建器使用Fluxtion流API描述资产计算。声明形式类似于Java流api，但是建立实时事件处理图。标有注释的方法
Maven插件调用@SepBuilder来生成静态事件处理器。以下代码描述了资产的计算，请参见
FluxtionBuilder ：

@SepBuilder(name = "SymbolTradeMonitor",packageName = "com.fluxtion.examples.tradingmonitor.generated.symbol",outputDir = "src/main/java",cleanOutputDir = true)public void buildAssetAnalyser(SEPConfig cfg) {//entry points subsrcibe to eventsWrapper<Deal> deals = select(Deal.class);Wrapper<AssetPrice> prices = select(AssetPrice.class);//result collector, and republish as an event sourceAssetTradePos results = cfg.addPublicNode(new AssetTradePos(), "assetTradePos");eventSource(results);//calculate derived valuesWrapper<Number> cashPosition = deals.map(multiply(), Deal::getSize, Deal::getPrice).map(multiply(), -1).map(cumSum());Wrapper<Number> pos = deals.map(cumSum(), Deal::getSize);Wrapper<Number> mtm = pos.map(multiply(), arg(prices, AssetPrice::getPrice));Wrapper<Number> pnl = add(mtm, cashPosition);//collect into resultscashPosition.push(results::setCashPos);pos.push(results::setAssetPos);mtm.push(results::setMtm);pnl.push(results::setPnl);deals.map(count()).push(results::setDealsProcessed);prices.map(count()).push(results::setPricesProcessed);//add some rules - only fires on first breachpnl.filter(lt(-200)).notifyOnChange(true).map(count()).push(results::setPnlBreaches);pos.filter(outsideBand(-200, 200)).notifyOnChange(true).map(count()).push(results::setPositionBreaches);//human readable names to nodes in generated code - not required deals.id("deals");prices.id("prices");cashPosition.id("cashPos");pos.id("assetPos");mtm.id("mtm");pnl.id("pnl");}

功能描述被转换为有效的命令形式以执行。生成的事件处理器SymbolTradeMonitor是AssetPrice和Deal事件的入口点。事件处理器使用生成的帮助程序类来计算聚合，这些帮助程序类在此处。

处理器从分区程序接收事件，并调用帮助程序函数以提取数据并调用计算函数，将聚合结果存储在节点中。汇总值被推送到结果实例AssetTradePos的字段中。不创建任何中间对象，无需自动装箱即可处理任何原始计算。计算节点从父实例引用数据，执行期间没有数据对象在图上移动。图形初始化后，处理事件时便没有内存分配。

与代码同时生成代表资产计算处理图的图像，如下所示：

FluxtionBuilderbuilder类的buildPortfolioAnalyser方法中描述了投资组合的一组类似计算，生成了PortfolioTradeMonitor事件处理程序。 AssetTradePos从SymbolTradeMonitor发布到PortfolioTradeMonitor。用于投资组合计算的生成文件位于此处。