介绍

在本文中，我将结合使用Lombok和Fluxtion这两种产品，以演示工具如何在减少代码编写和交付时间的同时提高代码的可读性。 使用Java 10中的var可以进一步改善这种情况。 产品和var都在构建时使用推断来加速开发。

Fluxtion的精神是最大程度地减少浪费，我们这里的目标是删除样板代码，减少代码噪声并简化集成任务。 我们希望花费尽可能少的开发时间，同时仍然提供能够每秒处理数百万条消息的高效且高性能的解决方案。

使用所描述的技术，我将Fluxtion / Lombok实现与使用Akka流的scala示例进行了比较，Java版本需要较少的代码，并且更易于构建。

家政服务，对未承认Richard Warburton的道歉
Opsian ，在我的第一个博客中 。

代码信噪比

在编写代码时，我们要解决两个主要任务：

• 将业务需求转换为程序逻辑
• 将逻辑与部署环境接口

理想情况下，我们希望将所有时间都花在第一个而不是第二个上。 另外，应减少编写的代码总量。 平衡抽象同时又要赋予开发人员权力并不容易，抽象太大了，我们就失去了表现力。 我希望能够与本文采用的方法取得良好的平衡。

想象一下，编写一些占用50行的税收计算逻辑，但是编写数据库，Web服务器，编组，日志记录等代码需要1000行。 尽管演示了技术能力，但在纯技术实施细节中并没有商业价值。 从另一个角度来看，我们可以将业务逻辑视为一种信号，将基础架构代码视为一种噪声。 我们编写的解决方案可以相对于有用的业务逻辑用信噪比来衡量。

维基百科将信噪比定义为：

信噪比（缩写为SNR或S / N）是用于
将所需信号的电平与背景噪声的电平进行比较的科学和工程学 。 SNR定义为信号功率与噪声功率之比，通常以分贝表示。 比率大于1：1（大于0 dB）表示信号多于噪声。

在大多数系统中，希望以高SNR比为目标，从编程的角度来看，高SNR的一些优点是：

• 减少编写代码
• 易于理解和维护的业务逻辑
• 学习曲线更短
• 调试/故障查找更简单，出错更少
• 更有效的发展

在Java中，多年来，从重量级的j2ee容器到更简单的框架（如spark和spring boot） ，我们一直感受到提高代码SNR的压力。 语言本身通过引入诸如lambda，流，方法引用和var变量声明之类的更改来适应了这一转变。

结合通量和Lombok

在该示例之前，先快速入门有关通量和Lombok。

助焊剂

Fluxtion是用Java编写的可嵌入流事件处理引擎。 开发人员以声明式和命令式的混合形式描述处理过程，以便Fluxtion可以生成决策引擎。 该引擎已序列化为Java代码，并且可以嵌入任何Java应用程序中。 该应用程序将事件输入引擎以进行流处理。

引擎生成可以在应用程序中内联发生，也可以通过maven插件在生成过程中进行。

Lombok底漆

Lombok是一种实用程序，可以自动为Java类编写样板代码，从而节省开发人员时间并减少代码噪音。 作为注释处理工具执行时，Lombok会生成表示已注释类的样板代码的字节代码。 Lombok功能集不完整包括：

• 属性的自动bean样式获取器和设置器
• 哈希代码和为属性生成的等于
• 自动toString方法
• 所有类属性的自动构造函数

只需将Lombok添加到您的Maven构建中，您的ide应该就可以使用，或者与netbeans和intellij一起使用。

流最大温度示例

让我们看一下常见的Fluxtion使用模式。 订阅事件流，从事件中提取值，对该值执行计算，过滤并将结果压入用户对象。 在这个简单的示例中，我们满足以下要求：

• 倾听温度事件
• 提取温度
• 保持最高温度
• 当出现新的最高温度时，将温度推入用户定义的实例中

克隆的回购从GitHub和使用本文的标记版本。 该项目在这里 。

git clone --branch  article_lombok_july2019 https://github.com/gregv12/articles.git

cd articles/2019/june/lombok/

满足处理要求的助焊剂代码：

select(TempEvent::getTemp).map(max()).notifyOnChange(true).push(new MyTempProcessor()::setMaxTemp);

这提供了较高的代码SNR和较低的行数，所有代码都是针对业务逻辑的。 要实现此功能，需要使用方法引用和类型推断。 方法引用允许Fluxtion推断所需的行为，要构建的功能，源和目标类型以及如何在执行图中将数据从一个节点传递到另一个节点。 方法参考为我们提供了一种表达任意逻辑的令人愉快的类型安全方式。 该工具采用的推论消除了开发人员的负担，以明确表达每个处理步骤，从而为我们提供了一个低代码环境。

在生成Fluxtion之后，序列化的流事件处理器为
在此 ，以Java代码表示。 示例的测试在这里 。

@Testpublic void testTemp() throws Exception{EventHandler handler = new InlineLombok().handler();((Lifecycle)handler).init();handler.onEvent(new InlineLombok.TempEvent(10));handler.onEvent(new InlineLombok.TempEvent(9));handler.onEvent(new InlineLombok.TempEvent(17));handler.onEvent(new InlineLombok.TempEvent(16));handler.onEvent(new InlineLombok.TempEvent(14));handler.onEvent(new InlineLombok.TempEvent(24));Assert.assertEquals(3, MyTempProcessor.count);}

输出：

08:08:42.921 [main] INFO  c.f.generator.compiler.SepCompiler - generated sep: D:\projects\fluxtion\articles\2019\june\lombok\target\generated-sources\fluxtion\com\fluxtion\articles\lombok\temperature\generated\lombok\TempMonitor.java
new max temp:10.0
new max temp:17.0
new max temp:24.0
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 4.79 sec

处理图形图像：

仔细观察上面示例中的第一行select（TempEvent :: getTemp） ，我们可以检查Fluxtion所做的推断。 这里暗示的逻辑是：

• 为TempEvent类型的事件创建订阅，
• 添加一个从传入事件中提取getTemp值的节点
• 使临时值可用作节点的Number属性
• 当接收到温度事件时，通知孩子改变温度值。

映射，notifyOnChange和push函数是添加到执行链中的步骤。 有关详细信息，请参见Fluxtion流模块的包装器接口。 由于SNR高，因此很容易理解它们的目的和效果，但出于完整性考虑：

• map（max（））从上一个节点（温度）提取number属性。 接收到新值时，将该值应用于有状态max函数。 将当前最大值存储在具有Number属性的节点中。 接收事件时，通知任何子节点当前最大值。
• notifyOnChange一个有状态函数，在监视的值已更新且与先前的值不同时触发。 仅新的最大值传播到子节点。
• push（new MyTempProcessor（）:: setMaxTemp）将用户节点MyTempProcessor添加到执行链中。 当由新的最大温度触发时，将节点的值推入MyTempProcessor的setMaxTemp中。 对原始类型执行所有类型转换，而不会产生垃圾。

要在TempEvent上使用方法引用，我们首先需要定义一个getter / setter样式访问器方法对。 当然，ide可以生成所需的方法，但是SNR在生成后仍会下降。 将其扩展到更大的域，问题成倍增加。 Lombok可以在这里为我们提供帮助，删除不必要的代码并恢复SNR。

在Lombok之前：

public class InlineNoLombok {public EventHandler handler() throws Exception {return sepInstance(c-> select(TempEvent::getTemp).map(max()).notifyOnChange(true).push(new MyTempProcessor()::setMaxTemp),"com.fluxtion.articles.lombok.temperature.generated.nolombok", "TempMonitor");}public static class TempEvent extends Event {private double temp;public TempEvent(double temp) {this.temp = temp;}public double getTemp() {return temp;}public void setTemp(double temp) {this.temp = temp;}}}

Lombok之后：

添加单个@Data批注将删除getter / setter，而@AllArgsConstructor则删除构造函数：

public class InlineLombok {public EventHandler handler() throws Exception {return sepInstance(c-> select(TempEvent::getTemp).map(max()).notifyOnChange(true).push(new MyTempProcessor()::setMaxTemp),"com.fluxtion.articles.lombok.temperature.generated.nolombok", "TempMonitor");}@Data@AllArgsConstructorpublic static class TempEvent extends Event {private double temp;}
}

即使使用Lombok和Fluxtion的最小示例，实际的业务逻辑也更易于阅读。 更好的代码SNR使应用程序的构建效率更高，更易于理解。

航班资料示例

让我们将其扩展到一个更复杂的示例，其中高SNR的值变得显而易见。 在此示例中，我们正在处理整个一年的航班数据。 该示例的灵感来自此博客 ，而akka流解决方案的代码在此处 。 要求摘要：

处理以CSV格式存储的一年的所有美国航班着陆记录的价值
在这里 。

• 按名称分组运营商
• 筛选延迟> 0的记录
• 运营商名称：第8列，延迟时间：第14列
• 对于运营商分组，计算：
  • 总延迟累计
• 计算航班总数，不考虑延误

我们需要定义数据类型和处理逻辑来解决该问题。 解决方案中的噪声很容易使您不知所措。 但是Fluxtion允许我们专注于业务逻辑，而Lombok使数据类型易于使用，两种工具都使用推理来减少编写代码：

public class FlightAnalyser {@SepBuilder(name = "FlightDelayAnalyser",packageName = "com.fluxtion.articles.lombok.flight.generated")public void buildFlightProcessor(SEPConfig cfg) {var flightDetails = csvMarshaller(FlightDetails.class, 1).map(14, FlightDetails::setDelay).converter(14, defaultInt(-1)).map(8, FlightDetails::setCarrier).converter(8, Converters::intern).build();//filter and group byvar delayedFlight = flightDetails.filter(FlightDetails::getDelay, positive());var carrierDelay = groupBy(delayedFlight, FlightDetails::getCarrier, CarrierDelay.class);//derived values for a groupcarrierDelay.init(FlightDetails::getCarrier, CarrierDelay::setCarrierId);carrierDelay.avg(FlightDetails::getDelay, CarrierDelay::setAvgDelay);carrierDelay.count(CarrierDelay::setTotalFlights);carrierDelay.sum(FlightDetails::getDelay, CarrierDelay::setTotalDelayMins);//make public for testingvar delayByGroup = cfg.addPublicNode(carrierDelay.build(), "delayMap");//dump to console, triggers on EofEventprintValues("\nFlight delay analysis\n========================",delayByGroup, eofTrigger());}@Data //input data from CSVpublic static class FlightDetails {private String carrier;private int delay;}@Data //derived datapublic static class CarrierDelay {private String carrierId;private int avgDelay;private int totalFlights;private int totalDelayMins;}}

实施分析

Lombok允许我们处理数据类和字段类型，而无需使用getter / setter方法。 我们定义输入记录FlightDetails和分组摘要记录CarrierDelay。

使用var关键字进行中间实例分配可以简化代码的读写。

• 第8行 Fluxtion将csv映射到FlightDetails类型，该数字1表示要忽略的初始标题行。
• 第9行将第 14列映射到延迟值。 可选的转换器函数将丢失的或非数字的延迟映射到值-1。 通过Fluxtion进行类型推断可确保从零gc到char到int的转换
• 第10行将第 8列映射到运营商名称。 可以使用载体名称，以减少不必要的String对象分配，因为我们希望相同的载体名称会多次出现。 请记住，有700万条记录将大大降低gc压力。
• 第12行 ，过滤器函数positive（）应用于字段FlightDetails :: getDelay。 只有延迟的航班由子节点处理。
• 第13行的已过滤记录delayFlight由关键字FlightDetails :: getCarrier分组，该组的目标是CarrierDelay。
• 第15行定义了新载波进入组的初始化函数，仅当在组中分配了新密钥时才调用。
• 第16行将平均值函数应用于延迟并设置值CarrierDelay：setAvgDelay
• 第17行将 count函数应用于延迟并设置值CarrierDelay：setTotalFlights
• 第18行将求和函数应用于延迟并设置值CarrierDelay：setTotalDelayMinutes

计算是有状态的，并且对于每个承运人都有唯一的值，每次收到FlightDelay记录时，相关承运人的计算都会更新。

• 第21行将delayMap分配为公共最终变量以协助测试
• 第22行在收到文件结束事件时打印映射值

性能

执行2008年的航班分析，解压缩Flight Csv数据并将文件位置传递到发行版中的可执行jar。

java.exe -jar dist\flightanalyser.jar [FLIGHT_CSV_DATA]

Flight delay analysis
========================
FlightAnalyser.CarrierDelay(carrierId=OO, avgDelay=31, totalFlights=219367, totalDelayMins=6884487)
FlightAnalyser.CarrierDelay(carrierId=AA, avgDelay=35, totalFlights=293277, totalDelayMins=10414936)
FlightAnalyser.CarrierDelay(carrierId=MQ, avgDelay=35, totalFlights=205765, totalDelayMins=7255602)
FlightAnalyser.CarrierDelay(carrierId=FL, avgDelay=31, totalFlights=117632, totalDelayMins=3661868)
FlightAnalyser.CarrierDelay(carrierId=DL, avgDelay=27, totalFlights=209018, totalDelayMins=5839658)
FlightAnalyser.CarrierDelay(carrierId=NW, avgDelay=28, totalFlights=158797, totalDelayMins=4482112)
FlightAnalyser.CarrierDelay(carrierId=UA, avgDelay=38, totalFlights=200470, totalDelayMins=7763908)
FlightAnalyser.CarrierDelay(carrierId=9E, avgDelay=32, totalFlights=90601, totalDelayMins=2907848)
FlightAnalyser.CarrierDelay(carrierId=CO, avgDelay=34, totalFlights=141680, totalDelayMins=4818397)
FlightAnalyser.CarrierDelay(carrierId=XE, avgDelay=36, totalFlights=162602, totalDelayMins=5989016)
FlightAnalyser.CarrierDelay(carrierId=AQ, avgDelay=12, totalFlights=1908, totalDelayMins=23174)
FlightAnalyser.CarrierDelay(carrierId=EV, avgDelay=35, totalFlights=122751, totalDelayMins=4402397)
FlightAnalyser.CarrierDelay(carrierId=AS, avgDelay=27, totalFlights=62241, totalDelayMins=1714954)
FlightAnalyser.CarrierDelay(carrierId=F9, avgDelay=21, totalFlights=46836, totalDelayMins=992044)
FlightAnalyser.CarrierDelay(carrierId=B6, avgDelay=42, totalFlights=83202, totalDelayMins=3559212)
FlightAnalyser.CarrierDelay(carrierId=WN, avgDelay=26, totalFlights=469518, totalDelayMins=12633319)
FlightAnalyser.CarrierDelay(carrierId=OH, avgDelay=34, totalFlights=96154, totalDelayMins=3291908)
FlightAnalyser.CarrierDelay(carrierId=HA, avgDelay=18, totalFlights=18736, totalDelayMins=342715)
FlightAnalyser.CarrierDelay(carrierId=YV, avgDelay=37, totalFlights=111004, totalDelayMins=4159465)
FlightAnalyser.CarrierDelay(carrierId=US, avgDelay=28, totalFlights=167945, totalDelayMins=4715728)millis:2682

加工性能分析：

file size           = 673 Mb

record count        = 7,009,728

比较这两种解决方案，我们观察到以下几点：

• Java版本使用的代码少于scala版本
• 通量消除了定义图的需要，而不仅仅是业务逻辑
• 手动构建图形是错误的根源
• Lombok使数据类型像scala case类一样简洁
• var减少代码膨胀
• 信噪比高，使代码更易于维护和理解
• Fluxtion易于运行，不需要安装服务器，只需编译即可。

比较性能数字很困难，Akka版本只花了一分钟时间来运行示例，但是我没有足够的Akka经验来验证这一点。 此外，这是一个古老的博客，因此情况可能会继续发展。

结论

我们着手证明，如果我们选择了一套不错的工具，那么Java可以成为事件流的简洁语言。 Lombok和Fluxtion完美地结合在一起，使处理逻辑的声明式定义既简单又类型安全。 var的使用使代码更具可读性，更易于编写。 所有这些的关键是推论，每个工具都推论出不同类型的行为，所有这些行为都使编码人员不必显式指定行为：

• var –类型推断
• Lombok–推断锅炉板的实施
• 通量–推断加工图

对于Fluxtion，我们比较Akka版本如何要求开发人员明确定义处理图。 这不适用于更大，更复杂的情况，并且将成为错误的来源。 更糟糕的是，业务逻辑被技术基础架构所掩盖，从而使维护成本在未来更加昂贵。

最后要指出的是，该解决方案的性能非常出色，每秒处理260万条记录且gc为零。 我希望您喜欢这份工作，并很想尝试Fluxtion和Lombok。

致谢

AllSimon在github上 ，他对Fluxtion的贡献使我开始尝试Lombok

翻译自: https://www.javacodegeeks.com/2019/07/easy-event-processing-with-var-lombok-and-fluxtion.html