Hazelcast JET目前是分布式计算框架领域的新成员。根据Hazelcast团队的说法,他们甚至比 Apache Spark 和 Apache Flink 更快。查看 基准 。让我们看看如何使用Hazelcast JET和 Spring Boot ,当然我会像往常一样构建一个简单的演示。
Hazelcast JET和Spring Boot依赖项设置:
<dependency> <groupId>com.hazelcast.jet</groupId> <artifactId>hazelcast-jet</artifactId> <version>0.3.2-SNAPSHOT</version> </dependency>
并且让我们尝试创建一个Hazelcast JET实例并最终将其关闭,以便验证我们是否能够正确启动Hazelcast JET:
<b>try</b> { JetInstance instance = Jet.newJetInstance(); } <b>finally</b> { Jet.shutdownAll(); }
不幸的是,Spring Boot应用程序内部的Hazelcast JET实例创建将崩溃:
java.lang.NoSuchFieldError: JET at com.hazelcast.jet.impl.config.XmlJetConfigBuilder.getXmlType(XmlJetConfigBuilder.java:128) ~[hazelcast-jet-0.3.2-SNAPSHOT.jar!/:0.3.2-SNAPSHOT] at com.hazelcast.config.AbstractXmlConfigHelper.getNamespaceType(AbstractXmlConfigHelper.java:136) ~[hazelcast-3.7.6.jar!/:3.7.6] at com.hazelcast.config.AbstractXmlConfigHelper.<init>(AbstractXmlConfigHelper.java:72) ~[hazelcast-3.7.6.jar!/:3.7.6] at com.hazelcast.config.AbstractConfigBuilder.<init>(AbstractConfigBuilder.java:59) ~[hazelcast-3.7.6.jar!/:3.7.6] at com.hazelcast.jet.impl.config.XmlJetConfigBuilder.<init>(XmlJetConfigBuilder.java:67) ~[hazelcast-jet-0.3.2-SNAPSHOT.jar!/:0.3.2-SNAPSHOT] at com.hazelcast.jet.impl.config.XmlJetConfigBuilder.getConfig(XmlJetConfigBuilder.java:80) ~[hazelcast-jet-0.3.2-SNAPSHOT.jar!/:0.3.2-SNAPSHOT]
好吧,不要责怪Hazelcast JET团队,只是Spring Boot已经内置了Hazelcast IMDG(Spring Boot 1.5.3版本为3.7.6),其中这个版本包含了hazelcast ConfigType枚举,但没有包含JET字段。就这样。解决此问题的方法是在您的pom中升级Hazelcast IMDG。对我有用的是将hazelcast.version属性设置为:
<properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> <hazelcast.version>3.8</hazelcast.version> </properties>
在这个解决方法之后,一切都应该工作正常,你应该能够看到他们漂亮的ascii艺术:-)
Hazelcast JET架构
在进入JET演示之前,让我们描述一下它的架构。Hazelcast JET基于将计算工作组合到 DAG图中 。这不是革命性的东西,因为Apache Spark也基于DAG工作。但是让我们更深入地了解Hazelcast JET DAG术语::
为了向您展示一个示例,我们可以构建一个简单的JET Job,它将 RabbitMQ代理中的 数据注入到分布式Hazelcast Map中。我们的工作将包含两个Vertex ,第一个Vertex 将发出来自RabbitMQ的项目,第二个Vertex 将是简单的Sink进入Map。
<b>private</b> <b>static</b> Job createJetJob(JetInstance instance) { DAG dag = <b>new</b> DAG(); Properties props = props( <font>"server"</font><font>, </font><font>"localhost"</font><font>, </font><font>"user"</font><font>, </font><font>"guest"</font><font>, </font><font>"password"</font><font>, </font><font>"guest"</font><font>); Vertex source = dag.newVertex(</font><font>"source"</font><font>, readRabbitMQ(props, </font><font>"jetInputQueue"</font><font>)); Vertex sink = dag.newVertex(</font><font>"sink"</font><font>, writeMap(</font><font>"sink"</font><font>)); dag.edge(between(source, sink)); <b>return</b> instance.newJob(dag); } </font>
详细代码
方法readRabbitMQ返回一个JET Processor实现,用于从RabbitMQ读取消息。该实现的灵感来自官方的 ReadKafkaP处理器 。我刚刚以轮询方式重写了连接器以与RabbitMQ代理进行通信。读取部分的消息在AbstractProcessor.complete方法中。要了解这种方法,您需要了解两件事:
轮询RabbitMQ消息的完整方法:
@Override <b>public</b> <b>boolean</b> complete() { System.out.println(<font>"....Invoking RabbitMQ vertex processor complete..."</font><font>); <b>if</b> (emitCooperatively(traverser)) { <b>final</b> Message message = <b>this</b>.rabbitTemplate.receive(<b>this</b>.queueNames[0]); <b>if</b> (message != <b>null</b>) { System.out.println(</font><font>"Message payload: "</font><font> + <b>new</b> String(message.getBody())); <b>final</b> List<Message> list = <b>new</b> ArrayList<>(); list.add(message); Random rn = <b>new</b> Random(); traverser = traverseStream(list.stream()).map(r -> entry(String.valueOf(rn.nextInt()), <b>new</b> String(r.getBody())) ); } } <b>return</b> false; } </font>
Hazelcast JET和工作分配到集群
将来自RabbitMQ代理的流式传输写入Hazelcast JET时,接收器Vertex 中的每个处理器都将成为消费者,从而完成相同的工作。
来自AMPQ队列的有效负载将在JET处理器之间进行负载平衡。无论如何,为特定Vertex 创建处理器的方式是 ProcessorMetaSupplier 和 ProcessorSupplier 实现的工作。
简而言之,JET使用ProcessorMetaSupplier实现为DAG图中的每个Vertex 获取ProcessorSupplier。然后将ProcessorSupplier发送到Vertex ,根据并行度设置创建处理器。我强烈建议您阅读JET文档中的 NumberGenerator 示例,这有助于了解JET如何创建处理器。
在RabbitMQ流媒体的情况下,我再说一遍,每个处理器都将做同样的工作,因此:
<b>private</b> <b>static</b> <b>final</b> <b>class</b> MetaSupplier<K, V> implements ProcessorMetaSupplier { <b>static</b> <b>final</b> <b>long</b> serialVersionUID = 1L; <b>private</b> <b>final</b> String[] queueNames; <b>private</b> Properties properties; <b>private</b> MetaSupplier(String[] queueNames, Properties properties) { <b>this</b>.queueNames = queueNames; <b>this</b>.properties = properties; } @Override <b>public</b> Function<Address, ProcessorSupplier> get(List<Address> addresses) { <b>return</b> address -> <b>new</b> Supplier<>(queueNames, properties); } } <b>private</b> <b>static</b> <b>class</b> Supplier<K, V> implements ProcessorSupplier { <b>static</b> <b>final</b> <b>long</b> serialVersionUID = 1L; <b>private</b> <b>final</b> String[] queueNames; <b>private</b> <b>final</b> Properties properties; <b>private</b> <b>transient</b> List<Processor> processors; Supplier(String[] topicIds, Properties properties) { <b>this</b>.properties = properties; <b>this</b>.queueNames = topicIds; } @Override <b>public</b> List<Processor> get(<b>int</b> count) { <b>return</b> processors = range(0, count) .mapToObj(i -> <b>new</b> ReadRabbitMQP<>(queueNames, properties)) .collect(toList()); } @Override <b>public</b> <b>void</b> complete(Throwable error) { processors.stream() .filter(p -> p instanceof ReadRabbitMQP) .map(p -> (ReadRabbitMQP) p) .forEach(p -> Util.uncheckRun(p::close)); } }
测试此演示
现在将一些消息发送到“jetInputQueue”输入4条以后输出:
Received 4 entries in 78080 milliseconds. ....Invoking RabbitMQ vertex processor complete... ....Invoking RabbitMQ vertex processor complete... Received 4 entries in 78182 milliseconds. ....Invoking RabbitMQ vertex processor complete... ....Invoking RabbitMQ vertex processor complete... Received 4 entries in 78283 milliseconds. ....Invoking RabbitMQ vertex processor complete... ....Invoking RabbitMQ vertex processor complete... Received 4 entries in 78385 milliseconds. ....Invoking RabbitMQ vertex processor complete... ....Invoking RabbitMQ vertex processor complete... Received 4 entries in 78486 milliseconds. ....Invoking RabbitMQ vertex processor complete..
我对Hazelcast JET的看法
优点:
缺点:
无论如何JET引起了我的注意,我期待着下一个版本。
点击标题看原文