后端并發(fā)編程異步編程【免費下載鏈接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.項目地址https://gitcode.com/gh_mirrors/ak/akka-core點擊查看免費下載Akka Streams 完整實現(xiàn)了 Reactive Streams 標(biāo)準(zhǔn)能夠以非阻塞背壓back pressure的方式與任何遵循該標(biāo)準(zhǔn)的流式庫互通。本篇指南基于官方文檔 reactive-streams-interop.md結(jié)合倉庫內(nèi) ReactiveStreamsDocSpec.scala 與 ReactiveStreamsDocTest.java 的真實測試示例以及Source、Sink、Flow工廠與JavaFlowSupport的底層源碼完整講解Publisher、Subscriber、Processor三類接口如何進(jìn)出 Akka Streams 圖幫助你把 Akka Streams 與 Reactor、RxJava 等第三方流式庫對接起來。讀完你將掌握Source 端接入外部 Publisher、Sink 端輸出外部 Subscriber、Flow 與 Processor 雙向轉(zhuǎn)換、fanout 廣播語義及背壓緩沖調(diào)優(yōu)。依賴配置Akka 官方組件發(fā)布在 Akka 的安全庫倉庫需要通過 https://account.akka.io/token 獲取帶 token 的安全 URL 訪問。使用 Akka Streams 需要先在項目中加入akka-stream模塊依賴官方推薦通過 BOMBill of Materials統(tǒng)一管理版本!-- Maven -- properties akka.version2.x.x/akka.version scala.binary.version2.13/scala.binary.version /properties dependencyManagement dependencies dependency groupIdcom.typesafe.akka/groupId artifactIdakka-bom_${scala.binary.version}/artifactId version${akka.version}/version typepom/type scopeimport/scope /dependency /dependencies /dependencyManagement dependency groupIdcom.typesafe.akka/groupId artifactIdakka-stream_${scala.binary.version}/artifactId /dependency// sbt val AkkaVersion 2.x.x libraryDependencies com.typesafe.akka %% akka-stream % AkkaVersion// Gradle def akkaVersion 2.x.x dependencies { implementation platform(com.typesafe.akka:akka-bom_2.13:${akkaVersion}) implementation com.typesafe.akka:akka-stream_2.13 }版本號請以你實際使用的 Akka 版本為準(zhǔn)當(dāng)前倉庫對應(yīng) Akka 2.9 系列$akka.version$占位符由文檔構(gòu)建時自動替換。概述Akka Streams 對 Reactive Streams 標(biāo)準(zhǔn)的實現(xiàn)Reactive Streams 是異步流處理與非阻塞背壓的行業(yè)標(biāo)準(zhǔn)。Akka Streams 實現(xiàn)了該標(biāo)準(zhǔn)因此可以把 Akka Streams 的圖與任何同樣遵循標(biāo)準(zhǔn)的流式庫自由拼接。在 API 形態(tài)上Reactive Streams 存在兩個版本org.reactivestreams包Java 8 時代的獨立 artifactorg.reactivestreams:reactive-streams定義了Publisher、Subscriber、Subscription、Processor四個接口java.util.concurrent.Flow包自 Java 9 起這四個接口被并入 JDK 標(biāo)準(zhǔn)庫位于java.util.concurrent.Flow命名空間下接口語義與org.reactivestreams完全一致。Akka Streams 對兩套接口都提供了互操作能力org.reactivestreams接口直接通過常規(guī)Source/SinkAPI 上的工廠方法接入即Source.fromPublisher、Sink.fromSubscriber、Sink.asPublisher、Source.asSubscriber、Flow.toProcessor、Flow.fromProcessorJava 9 內(nèi)置的java.util.concurrent.Flow.*接口則通過獨立的工廠集合接入Scala 端為akka.stream.scaladsl.JavaFlowSupportJava 端為akka.stream.javadsl.JavaFlowSupport。下文示例均以org.reactivestreams工廠演示每一處調(diào)用都可以原樣替換為JavaFlowSupport中的對應(yīng)方法并使用 JDK 的java.util.concurrent.Flow.*接口。需要特別注意的是JavaFlowSupport無法在 Java 8 上使用因為所需接口不在 JDK 8 標(biāo)準(zhǔn)庫中這一點在兩個JavaFlowSupport的源碼注釋中也明確標(biāo)注了 For use only with JDK 9參見 scaladsl/JavaFlowSupport.scala 與 javadsl/JavaFlowSupport.java。核心接口與一個完整的端到端示例Reactive Streams 中最重要的是兩個接口Publisher數(shù)據(jù)發(fā)布者向下游推送元素并響應(yīng)需求信號和Subscriber數(shù)據(jù)訂閱者向上游提交需求并接收元素。先看導(dǎo)入Scala來自 ReactiveStreamsDocSpec.scalaimport org.reactivestreams.Publisher import org.reactivestreams.Subscriber import org.reactivestreams.ProcessorJava來自 ReactiveStreamsDocTest.javaimport org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Processor;假設(shè)某個第三方庫對外提供了一個推文的Publisher// Scalalibrary provides a publisher of tweets def tweets: Publisher[Tweet]// Java PublisherTweet tweets();另一個庫知道如何把作者賬號存進(jìn)數(shù)據(jù)庫它接收一個Subscriber[Author]// Scala def storage: Subscriber[Author]// Java SubscriberAuthor storage();現(xiàn)在用 Akka Streams 的Flow做變換并把兩者連接起來——先過濾出帶有 Akka 標(biāo)簽的推文再取出作者// Scala過濾 取作者 val authors Flow[Tweet].filter(_.hashtags.contains(akkaTag)).map(_.author) // 連接Publisher 作為 SourceSubscriber 作為 Sink Source.fromPublisher(tweets).via(authors).to(Sink.fromSubscriber(storage)).run()// Java final FlowTweet, Author, NotUsed authors Flow.of(Tweet.class).filter(t - t.hashtags().contains(AKKA)).map(t - t.author); Source.fromPublisher(rs.tweets()).via(authors).to(Sink.fromSubscriber(rs.storage()));這里可以看到最核心的互操作模式Source.fromPublisher(publisher)把外部Publisher當(dāng)作流的輸入SourceSink.fromSubscriber(subscriber)把外部Subscriber當(dāng)作流的輸出Sink。Flow夾在中間完成元素變換與背壓傳遞。該示例在測試中的完整斷言ReactiveStreamsDocSpec.scala驗證了 7 位作者按序到達(dá)并最終收到完成信號證明背壓與完成傳播都工作正常。將 Flow 轉(zhuǎn)換為 Processor一個Flow還可以被轉(zhuǎn)換為RunnableGraph[Processor[In, Out]]當(dāng)調(diào)用run()時物化出一個Processor實例。run()可以被多次調(diào)用每次都會得到新的Processor實例即可重用// Scala val processor: Processor[Tweet, Author] authors.toProcessor.run() tweets.subscribe(processor) // 外部 Publisher 訂閱 Processor 的輸入端 processor.subscribe(storage) // Processor 的輸出端被 Subscriber 訂閱// Java final ProcessorTweet, Author processor authors.toProcessor().run(system); rs.tweets().subscribe(processor); processor.subscribe(rs.storage());Processor既是Publisher又是Subscriber所以可以直接用subscribe方法把它接到兩端的標(biāo)準(zhǔn)接口上。其底層實現(xiàn)詳見后文“源碼級實現(xiàn)解析”正是Source.asSubscriberFlowSink.asPublisher(false)的拼接物化。將 Source 暴露為 Publisher把Source暴露給外部世界使用的是Publisher-Sink即Sink.asPublisher// Scala val authorPublisher: Publisher[Author] Source.fromPublisher(tweets).via(authors).runWith(Sink.asPublisher(fanout false)) authorPublisher.subscribe(storage)// Java final PublisherAuthor authorPublisher Source.fromPublisher(rs.tweets()) .via(authors) .runWith(Sink.asPublisher(AsPublisher.WITHOUT_FANOUT), system); authorPublisher.subscribe(rs.storage());注意runWith的物化值就是這個Publisher拿到后即可把它交給任何 Reactive Streams 兼容的消費者。單訂閱者fanout false使用Sink.asPublisher(fanout false)Java 端為AsPublisher.WITHOUT_FANOUT創(chuàng)建的Publisher只支持一次訂閱。后續(xù)再嘗試訂閱會被拒絕并拋出IllegalStateException。這一點在兩個JavaFlowSupport的Sink.asPublisher文檔注釋中均有明確說明scaladsl/JavaFlowSupport.scala、javadsl/JavaFlowSupport.java。多訂閱者fanout true廣播如果需要支持多個訂閱者就要使用 fan-out / 廣播模式// Scala一個 Publisher 同時服務(wù) storage 與 alert 兩個 Subscriber def alert: Subscriber[Author] val authorPublisher: Publisher[Author] Source.fromPublisher(tweets).via(authors).runWith(Sink.asPublisher(fanout true)) authorPublisher.subscribe(storage) authorPublisher.subscribe(alert)// Java final PublisherAuthor authorPublisher Source.fromPublisher(rs.tweets()) .via(authors) .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), system); authorPublisher.subscribe(rs.storage()); authorPublisher.subscribe(rs.alert());fanout trueAsPublisher.WITH_FANOUT下物化出的Publisher會向每個訂閱者廣播同樣的元素流。測試 ReactiveStreamsDocSpec.scala 中兩個訂閱者都完整收到了全部作者且斷言注釋特別說明“該測試依賴 fanout Publisher 的緩沖區(qū)大小大于作者數(shù)量”。fanout 的背壓語義與 input-buffer該運算符的輸入緩沖區(qū)大小input buffer size決定了最快訂閱者與最慢訂閱者之間最多能拉開多大差距超出后整個流會被迫減速。也就是說fanout true的Publisher會以最慢的訂閱者為節(jié)拍器快的訂閱者最多領(lǐng)先慢的訂閱者input-buffer個元素隨后上游處理被背壓抑制。若測試中的作者數(shù)量大于緩沖大小慢訂閱者就會因緩沖區(qū)溢出而丟數(shù)據(jù)或報錯——這正是測試注釋的用意。將 Sink 暴露為 Subscriber反過來用 Subscriber-SourceSource.asSubscriber可以把一個Sink暴露為外部可訂閱的Subscriber// Scala val tweetSubscriber: Subscriber[Tweet] authors.to(Sink.fromSubscriber(storage)).runWith(Source.asSubscriber[Tweet]) tweets.subscribe(tweetSubscriber)// Java final SubscriberTweet tweetSubscriber authors.to(Sink.fromSubscriber(storage)).runWith(Source.asSubscriber(), system); rs.tweets().subscribe(tweetSubscriber);這里Source.asSubscriber[Tweet]的物化值是一個Subscriber[Tweet]外部庫可以把整個 Akka Streams 圖當(dāng)作一個“會消費推文的 Subscriber”直接subscribe。Source.asSubscriber的實現(xiàn)定義于 Source.scala包裝內(nèi)部SubscriberSource階段是“Sink 端接收外部訂閱”的鏡像操作。將 Processor 重新包裝為 FlowProcessor實例也可以重新包裝回Flow但必須傳入一個創(chuàng)建 Processor 的工廠函數(shù)而非直接傳入 Processor 實例。只有工廠才能保證結(jié)果Flow的可重用性——因為一個Processor只能被物化使用一次而工廠可以在每次物化時生成新實例// Scala // An example Processor factory def createProcessor: Processor[Int, Int] Flow[Int].toProcessor.run() val flow: Flow[Int, Int, NotUsed] Flow.fromProcessor(() createProcessor)// Java final CreatorProcessorInteger, Integer factory new CreatorProcessorInteger, Integer() { public ProcessorInteger, Integer create() { return Flow.of(Integer.class).toProcessor().run(system); } }; final FlowInteger, Integer, NotUsed flow Flow.fromProcessor(factory);如果你需要保留物化值可以使用Flow.fromProcessorMat它接受返回(Processor, Mat)二元組的工廠函數(shù)Scala 端定義見 Flow.scala。源碼級實現(xiàn)解析以上工廠并非黑盒魔法它們的底層實現(xiàn)在倉庫里清晰可見理解后能幫你判斷背壓行為與資源開銷。1.Sink.asPublisher的兩個分支Scaladsl 的 Sink.asPublisher 根據(jù)fanout選擇兩個不同的內(nèi)部階段def asPublisherT: Sink[T, Publisher[T]] fromGraph( if (fanout) new FanoutPublisherSinkT) else new PublisherSinkT))在 impl/Sinks.scala 中可以看到兩者的本質(zhì)區(qū)別PublisherSink非 fanout創(chuàng)建一個VirtualPublisher作為物化值。此時流上還沒有任何訂閱者因此流在預(yù)取元素填滿內(nèi)部緩沖區(qū)后會一直保持背壓直到訂閱者接入并產(chǎn)生需求。源碼注釋還說明它同時注冊了StreamSubscriptionTimeout用于處理“流已啟動但遲遲無人訂閱”的超時場景FanoutPublisherSinkfanout直接啟動一個名為FanoutProcessorImpl的 actor通過context.materializer.actorOf(...)用ActorProcessor實現(xiàn)廣播分發(fā)這正是它天然支持多訂閱者的原因也是它具備獨立輸入緩沖區(qū)語義的根源。2.Flow.toProcessor的組裝邏輯Flow.scala 中的 toProcessor 內(nèi)部就是把Source.asSubscriber、Flow、Sink.asPublisher(false)串成一個圖物化后把Subscriber端與Publisher端打包成一個匿名Processordef toProcessor: RunnableGraph[Processor[In uncheckedVariance, Out uncheckedVariance]] Source .asSubscriber[In] .via(this) .toMat(Sink.asPublisherOut)(Keep.both[Subscriber[In], Publisher[Out]]) .mapMaterializedValue { case (sub, pub) new Processor[In, Out] { override def onError(t: Throwable): Unit sub.onError(t) override def onSubscribe(s: Subscription): Unit sub.onSubscribe(s) override def onComplete(): Unit sub.onComplete() override def onNext(t: In): Unit sub.onNext(t) override def subscribe(s: Subscriber[_ : Out]): Unit pub.subscribe(s) } }javadsl.JavaFlowSupport.Flow.toProcessor做了同樣的事但面向java.util.concurrent.Flow.Processorjavadsl/JavaFlowSupport.java。注意它固定使用WITHOUT_FANOUT因此toProcessor產(chǎn)出的Processor輸出端同樣只接受單個訂閱者。3.Flow.fromProcessor與ProcessorModuleFlow.fromProcessor最終落到fromGraph(ProcessorModule(processorFactory))Flow.scala。ProcessorModule定義在 impl/StreamLayout.scala它把“按需調(diào)用工廠創(chuàng)建 Processor”的語義固化為一個圖階段由 PhasedFusingActorMaterializer.scala 中的ProcessorModulePhase負(fù)責(zé)物化——每次物化都調(diào)用一次工廠這正是“Flow 可重用”的保障。4.JavaFlowSupport的適配轉(zhuǎn)換Java 9 的java.util.concurrent.Flow.*接口并非另起爐灶而是通過 impl/JavaFlowAndRsConverters 提供的asRs/asJava隱式轉(zhuǎn)換在Flow接口與org.reactivestreams接口之間做零成本適配。例如// scaladsl/JavaFlowSupport.scala def fromPublisherT: Source[T, NotUsed] scaladsl.Source.fromPublisher(publisher.asRs) def asSubscriber[T]: Source[T, java.util.concurrent.Flow.Subscriber[T]] scaladsl.Source.asSubscriber[T].mapMaterializedValue(_.asJava)也就是說JavaFlowSupport只是org.reactivestreams工廠的適配殼Scala 端見 scaladsl/JavaFlowSupport.scalaJava 端見 javadsl/JavaFlowSupport.java其運行時行為與對應(yīng)org.reactivestreams工廠完全一致。倉庫還提供了 JavaFlowSupportCompileTest.java 編譯級測試來保證這層適配的 API 完整性。與其他 Reactive Streams 實現(xiàn)互操作實現(xiàn) Reactive Streams 標(biāo)準(zhǔn)的價值在于Akka Streams 可以與同樣遵循該標(biāo)準(zhǔn)的其他流式庫直接拼接無需膠水代碼。官方文檔列舉的部分其他實現(xiàn)包括Reactor1.1RxJava通過 RxJavaReactiveStreams 橋接RatpackSlick無論對接哪種庫交互方式都一致對方提供Publisher就用Source.fromPublisher接進(jìn)來對方需要Subscriber就用Sink.fromSubscriber送出去需要雙向能力就用Flow.toProcessor/Flow.fromProcessor。這也意味著你在 Akka Streams 圖內(nèi)享受到的全部能力背壓、扇出/扇入、異步邊界、故障恢復(fù)等在跨越庫邊界后依然由 Reactive Streams 的信號協(xié)議完整承載。小結(jié)場景工廠方法物化值外部 Publisher 進(jìn)圖Source.fromPublisher(pub)NotUsed外部 Subscriber 出圖Sink.fromSubscriber(sub)NotUsedSource 暴露為 Publisher單訂閱Sink.asPublisher(fanout false)Publisher[T]Source 暴露為 Publisher廣播Sink.asPublisher(fanout true)Publisher[T]Sink 暴露為 SubscriberSource.asSubscriber[T]Subscriber[T]Flow 轉(zhuǎn)為 Processor可多次 runflow.toProcessor.run()Processor[In, Out]Processor 包回 Flow需工廠Flow.fromProcessor(() proc)NotUsed對于 Java 9 用戶將上表所有方法替換為JavaFlowSupport中的對應(yīng)版本、接口換成java.util.concurrent.Flow.*即可獲得完全等價的互操作能力。完整的可運行示例可繼續(xù)閱讀 ReactiveStreamsDocSpec.scala 與 ReactiveStreamsDocTest.java。贊分享后端并發(fā)編程異步編程【免費下載鏈接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.項目地址https://gitcode.com/gh_mirrors/ak/akka-core點擊查看免費下載相關(guān)推薦Akka Streams Sink.asPublisher 完全指南將 Akka Stream 橋接到 Reactive Streams PublisherAkka Streams Sink.asPublisher 完全指南將 Akka Stream 橋接到 Reactive Streams Publisher后端并發(fā)編程異步編程Reactive Streams四大核心組件詳解Publisher、Subscriber、Subscription、ProcessorReactive Streams四大核心組件詳解Publisher、Subscriber、Subscription、Processor Reactive St后端Akka Streams 設(shè)計原理從不可變藍(lán)圖到 Reactive Streams 互操作Akka Streams 設(shè)計原理從不可變藍(lán)圖到 Reactive Streams 互操作 Akka Streams 是 Akka 平臺中面向分布式有界流處理后端并發(fā)編程異步編程創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考