react api_使用React流API将Akka流与rxJava结合在一起

react api

这次只是快速的一篇文章，因为我仍在尝试这种东西。关于React式编程有很多话题。在Java 8中，我们有Stream API，有rxJava我们有ratpack ，Akka有akka-streams 。

这些实现的主要问题是它们不兼容。您不能将一个实现的订阅者连接到另一个实现的发布者。幸运的是，一项倡议已经开始提供一种方法，使这些不同的实现可以协同工作：

“本规范旨在允许创建许多符合标准的实现，这些实现将通过遵守规则将能够顺利互操作，并在流应用程序的整个处理图中保留上述优势和特征。”

来自– http://www.reactive-streams.org/

这是如何运作的

现在我们该怎么做？让我们看一下基于akka-stream提供的示例的快速示例（从此处开始）。在下面的清单中：

package sample.streamimport akka.actor.ActorSystem
import akka.stream.FlowMaterializer
import akka.stream.scaladsl.{SubscriberSink, PublisherSource, Source}
import com.google.common.collect.{DiscreteDomain, ContiguousSet}
import rx.RxReactiveStreams
import rx.Observable;
import scala.collection.JavaConverters._object BasicTransformation {def main(args: Array[String]): Unit = {// define an implicit actorsystem and import the implicit dispatcherimplicit val system = ActorSystem("Sys")import system.dispatcher// flow materializer determines how the stream is realized.// this time as a flow between actors.implicit val materializer = FlowMaterializer()// input text for the stream.val text ="""|Lorem Ipsum is simply dummy text of the printing and typesetting industry.|Lorem Ipsum has been the industry's standard dummy text ever since the 1500s, |when an unknown printer took a galley of type and scrambled it to make a type |specimen book.""".stripMargin// create an observable from a simple list (this is in rxjava style)val first = Observable.from(text.split("\\s").toList.asJava);// convert the rxJava observable to a publisherval publisher = RxReactiveStreams.toPublisher(first);// based on the publisher create an akka sourceval source = PublisherSource(publisher);// now use the akka style syntax to stream the data from the source// to the sink (in this case this is println)source.map(_.toUpperCase).                 // executed as actorsfilter(_.length > 3).foreach { el =>                     // the sink/consumerprintln(el)}.onComplete(_ => system.shutdown())  // lifecycle event}
}

此示例中的代码注释几乎解释了正在发生的事情。我们在这里所做的是创建一个基于rxJava的Observable。将此Observable转换为“React流”发布者，并使用此发布者创建akka-streams源。对于其余的代码，我们可以使用akka-stream样式流API对流进行建模。在这种情况下，我们只需要进行一些过滤并打印出结果即可。