后端并發(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點擊查看免費下載導(dǎo)讀Source.future是 Akka Streams 中用于把 Scala 標(biāo)準(zhǔn)庫Future[T]無縫接入流式處理管線的核心算子當(dāng)Future成功完成且下游產(chǎn)生需求demand時它會將唯一的元素發(fā)射給下游然后完成數(shù)據(jù)流。本文以官方文檔 Source.future 為主體結(jié)合 Akka 源碼中的實際實現(xiàn)Source.scala 與 GraphStages.scala和單元測試SourceSpec.scala講清其簽名、語義、底層原理、使用示例與邊界情況。讀完本文你將掌握如何把異步計算結(jié)果如 RPC 響應(yīng)、數(shù)據(jù)庫查詢結(jié)果以背壓安全的方式注入 Akka Streams 管線并能理解其與Source.completionStage、Source.futureSource等相鄰算子的異同。簽名SignatureSource.future定義在akka.stream.scaladsl.Source伴生對象中其 Scala 簽名為def futureT: Source[T, NotUsed]輸入一個scala.concurrent.Future[T]即 Scala 標(biāo)準(zhǔn)庫中表示將來某個時刻會完成的一個異步值的類型。輸出一個Source[T, NotUsed]即至多發(fā)射一個T類型元素的流式數(shù)據(jù)源其物化值materialized value為NotUsed表示不產(chǎn)生有意義的物化結(jié)果。Java 使用者無需直接使用該算子——Akka Streams 為 Java 標(biāo)準(zhǔn)庫的CompletionStage提供了對應(yīng)的Source.completionStage二者在語義上完全等價completionStage內(nèi)部就是通過completionStage.asScala轉(zhuǎn)成Future后調(diào)用future實現(xiàn)的見 Source.scala。描述DescriptionSource.future的核心語義只有一句話當(dāng)Future完成且下游存在需求時發(fā)射該Future的唯一值若Future以失敗結(jié)束則整個數(shù)據(jù)流以該異常失敗fail。這一語義意味著單元素?zé)o論Future何時完成數(shù)據(jù)流中最多只有一個元素。需求驅(qū)動元素只會在下游發(fā)出 pull 請求有需求時才被發(fā)射因此天然遵守 Reactive Streams 的背壓backpressure協(xié)議。失敗傳播Future的失敗會直接轉(zhuǎn)化為流的失敗下游所有算子與 Sink 都會收到該失敗信號這與 Akka Streams 中錯誤總是隨流傳播的總體設(shè)計一致。需要特別指出與Source.completionStage不同其文檔明確說明若CompletionStage以null完成則流不發(fā)射任何值直接完成Source.future對null的處理也遵循同一行為——從 GraphStages.scala 的實現(xiàn)可以看到Success(null)時調(diào)用completeStage()即不發(fā)射元素、直接完成流。Reactive Streams 語義官方文檔用如下兩條規(guī)則概括該算子的流語義項目語義emits發(fā)射當(dāng)Future完成時且下游有需求completes完成在Future完成之后示例Example官方文檔在 SourceOperators.scala 中給出了完整可運行的 Scala 示例以下即為#sourceFromFuture代碼片段import akka.stream.scaladsl._ import akka.{ Done, NotUsed } import scala.concurrent.Future val source: Source[Int, NotUsed] Source.future(Future.successful(10)) val sink: Sink[Int, Future[Done]] Sink.foreach((i: Int) println(i)) val done: Future[Done] source.runWith(sink) // 輸出: 10解讀這段代碼的執(zhí)行流程Future.successful(10)創(chuàng)建一個已經(jīng)成功完成、值為10的FutureSource.future(...)將其包裝成單元素SourceSink.foreach對每個到達(dá)的元素執(zhí)行println其物化值為Future[Done]可用于等待流完成runWith將流連接并物化控制臺輸出10最終done會在流完成后完成。需要implicit的ActorSystem示例代碼中為implicit val system: ActorSystem ???來物化流。若希望在真實應(yīng)用中等待結(jié)果可以Await.result(done, timeout)或繼續(xù)在done上鏈?zhǔn)浇M合后續(xù)邏輯。對應(yīng)的 Java 版本使用CompletionStage可以在 FromCompletionStage.java 中看到import akka.Done; import akka.NotUsed; import akka.stream.javadsl.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; CompletionStageInteger stage CompletableFuture.completedFuture(10); SourceInteger, NotUsed source Source.completionStage(stage); SinkInteger, CompletionStageDone sink Sink.foreach(i - System.out.println(i.toString())); source.runWith(sink, system); // 輸出: 10底層實現(xiàn)與原理源碼級剖析Source.future的快路徑與常規(guī)路徑分界清晰理解它有助于你判斷不同場景下的性能特征。工廠方法的快速路徑優(yōu)化在 Source.scala 中future工廠方法會先檢查futureElement.valueFuture的瞬時狀態(tài)快照def futureT: Source[T, NotUsed] futureElement.value match { case Some(Success(null)) empty // 已完成且值為 null → 空流 case Some(Success(value)) single(value) // 已完成且值非 null → 單元素流 case Some(Failure(cause)) failed(cause) // 已失敗 → 直接失敗流 case None fromGraph(new FutureSourceT) // 未完成 → 通用實現(xiàn) }也就是說如果Future在調(diào)用Source.future時已經(jīng)完成工廠方法會直接退化為更輕量的Source.empty、Source.single或Source.failed繞開通用的FutureSource圖階段GraphStage從而避免不必要的異步回調(diào)開銷。這一點在 SourceSpec.scala 的測試中得到驗證已成功的Future會被優(yōu)化為singleSource已失敗的Future會被優(yōu)化為failedSource未完成的Future用Promise表示才會使用futureSource。三個測試均通過source.getAttributes.nameLifted斷言了優(yōu)化后的屬性名并驗證了運行結(jié)果成功值done、失敗異常TE(boom)。未完成 Future 的通用實現(xiàn)FutureSource GraphStage當(dāng)Future尚未完成時Source.future會構(gòu)造一個內(nèi)部InternalApi圖階段FutureSource完整實現(xiàn)位于 GraphStages.scala。其關(guān)鍵邏輯如下final class FutureSourceT extends GraphStage[SourceShape[T]] { val shape SourceShape(OutletT) val out shape.out override def createLogic(attr: Attributes) new GraphStageLogic(shape) with OutHandler { def onPull(): Unit { future.value match { case Some(completed) // optimization if the future is already completed onFutureCompleted(completed) case None val cb getAsyncCallback[Try[T]](onFutureCompleted).invoke _ future.onComplete(cb)(ExecutionContext.parasitic) } def onFutureCompleted(result: Try[T]): Unit { result match { case scala.util.Success(null) completeStage() case scala.util.Success(v) emit(out, v, () completeStage()) case scala.util.Failure(t) failStage(t) } } setHandler(out, eagerTerminateOutput) // After first pull we wont produce anything more } setHandler(out, this) } }其運作機制體現(xiàn)了 Akka Streams 圖階段GraphStage設(shè)計的幾個要點拉取觸發(fā)onPull()在首次收到下游需求時被調(diào)用。此時先做一次future.value快照檢查——若Future已在此期間完成則直接走onFutureCompleted同步完成無需注冊回調(diào)否則注冊回調(diào)。線程安全回調(diào)Future.onComplete的回調(diào)發(fā)生在線程池線程上而 GraphStage 邏輯必須運行在流引擎的專用線程上因此這里用getAsyncCallback將回調(diào)結(jié)果安全地投遞回流的執(zhí)行上下文規(guī)避了任何競態(tài)與線程安全問題。惰性注冊回調(diào)只在第一次onPull時才注冊因此如果下游從未請求元素例如下游提前取消則不會為Future注冊回調(diào)體現(xiàn)了按需demand-driven的資源使用策略。一次性發(fā)射發(fā)射元素后通過emit(out, v, () completeStage())在發(fā)射完成時立即結(jié)束流在第一次 pull 之后handler 被替換為eagerTerminateOutput保證不會再有第二次發(fā)射。失敗傳播Failure(t)分支調(diào)用failStage(t)將異常作為流失敗傳播給下游。這與文檔若Future失敗則流以該異常失敗的描述一一對應(yīng)。與其他相關(guān)算子的對比算子輸入輸出元素物化值說明Source.futureFuture[T]至多 1 個TNotUsed本文主角Scala 原生Source.completionStageCompletionStage[T]至多 1 個TNotUsedJava 互操作版本內(nèi)部委托給future見 Source.scalaSource.futureSourceFuture[Source[T, M]]多個T來自內(nèi)部 SourceFuture[M]等Future完成后展開為內(nèi)層Source的全部元素見 Source.scalaSource.lazyFutureSource.lazilyAsync的替代() Future[T]工廠至多 1 個TFuture[NotUsed]工廠延遲到下游產(chǎn)生需求時才調(diào)用見 Source.scala 的廢棄注釋選用建議已經(jīng)持有或即將持有一個Future計算結(jié)果且只需要把它作為流中唯一元素繼續(xù)加工時用Source.future在 Java 代碼中使用Source.completionStage當(dāng)Future本身展開后是一個Source例如異步獲取一個流式查詢結(jié)果時用Source.futureSource當(dāng)希望延遲創(chuàng)建Future即下游沒有需求就完全不發(fā)起異步計算時用Source.lazyFuture或Source.lazySource。邊界情況與注意事項null值Future成功完成但值為null時流不發(fā)射任何元素直接完成等價于Source.empty。失敗傳播Future失敗會導(dǎo)致流失敗異常沿流向下傳播請確保流中后續(xù)的算子/接收方對失敗有兜底處理如Sink.onComplete、recover等。取消行為若下游在Future完成前取消回調(diào)不會被注冊Future的完成結(jié)果將無人消費——若該Future承載著昂貴資源如數(shù)據(jù)庫連接需在業(yè)務(wù)側(cè)自行處理取消語義。物化值NotUsed意味著該算子本身不提供等待完成的句柄若要等整個流結(jié)束請依賴runWith(sink)返回的 Sink 物化值如示例中的Future[Done]。已完成的FutureSource.future(Future.successful(v))會被直接優(yōu)化為Source.single(v)性能上無需擔(dān)憂同理Future.failed(e)會被優(yōu)化為Source.failed(e)?？偨Y(jié)Source.future是 Akka Streams 中異步計算 → 流轉(zhuǎn)換的基礎(chǔ)構(gòu)件它以背壓安全的方式把 ScalaFuture的結(jié)果變成流中的唯一元素將失敗透明地轉(zhuǎn)化為流失敗并對已完成的Future提供了零開銷的快路徑優(yōu)化。結(jié)合 官方文檔、工廠實現(xiàn)、FutureSource GraphStage 實現(xiàn) 與 單元測試你可以放心地將它用于 RPC 響應(yīng)、緩存讀取、數(shù)據(jù)庫查詢等典型異步場景并與其他Source算子自由組合構(gòu)建完整的流式處理管線。若使用 Java請使用其等價物Source.completionStage見 completionStage.md。贊分享后端并發(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 的 Source.futureSource 算子將異步 Future[Source] 轉(zhuǎn)換為流式數(shù)據(jù)源Akka Streams 的 Source.futureSource 算子將異步 Future Source 轉(zhuǎn)換為流式數(shù)據(jù)源 本篇文章深入剖析 Akka S后端并發(fā)編程異步編程Akka Streams Source.lazyFuture 操作符完全指南延遲創(chuàng)建單元素 Future 的惰性數(shù)據(jù)源Akka Streams Source.lazyFuture 操作符完全指南延遲創(chuàng)建單元素 Future 的惰性數(shù)據(jù)源 本文圍繞 Akka 項目中 Sourc后端并發(fā)編程異步編程Akka Streams Sink.seq 算子詳解把流中的元素收集為集合Akka Streams Sink.seq 算子詳解把流中的元素收集為集合 Sink.seq 是 Akka Streams 中一個常用且易用的下游算子Sin后端并發(fā)編程異步編程創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考