這次只是一篇簡(jiǎn)短的文章，因?yàn)槲胰栽趪L試這種東西。 關(guān)于反應(yīng)式編程有很多話(huà)題。 在Java 8中，我們有Stream API，我們有rxJava我們有ratpack ，Akka有akka-streams 。

這些實(shí)現(xiàn)的主要問(wèn)題是它們不兼容。 您不能將一個(gè)實(shí)現(xiàn)的訂閱者連接到另一個(gè)實(shí)現(xiàn)的發(fā)布者。 幸運(yùn)的是，一項(xiàng)倡議已經(jīng)開(kāi)始提供一種方法，使這些不同的實(shí)現(xiàn)可以協(xié)同工作：

“本規(guī)范旨在允許創(chuàng)建許多符合標(biāo)準(zhǔn)的實(shí)現(xiàn)，這些實(shí)現(xiàn)將通過(guò)遵守規(guī)則將能夠平滑地互操作，并在流應(yīng)用程序的整個(gè)處理圖中保留上述好處和特征。”

來(lái)自– http://www.reactive-streams.org/

這是如何運(yùn)作的

現(xiàn)在我們?cè)撛趺醋?#xff1f; 讓我們看一下基于akka-stream提供的示例的快速示例（從此處開(kāi)始 ）。 在下面的清單中：

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} }

此示例中的代碼注釋幾乎解釋了正在發(fā)生的事情。 我們?cè)谶@里所做的是創(chuàng)建一個(gè)基于rxJava的Observable。 將此Observable轉(zhuǎn)換為“反應(yīng)流”發(fā)布者，并使用此發(fā)布者創(chuàng)建akka-streams源。 對(duì)于其余的代碼，我們可以使用akka-stream樣式流API對(duì)流進(jìn)行建模。 在這種情況下，我們只需要進(jìn)行一些過(guò)濾并打印出結(jié)果即可。

翻譯自: https://www.javacodegeeks.com/2014/11/use-reactive-streams-api-to-combine-akka-streams-with-rxjava.html

總結(jié)

以上是生活随笔為你收集整理的使用反应流API将Akka流与rxJava结合在一起的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。