后端并發(fā)編程異步編程【免費(fèi)下載鏈接】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點(diǎn)擊查看免費(fèi)下載本篇文章以 Akka 官方文檔 akka-docs/src/main/paradox/stream/operators/Source/repeat.md 為主體結(jié)合 akka-stream 模塊的源碼與測試用例系統(tǒng)講解Source.repeat操作符的簽名、語義、底層實現(xiàn)、Reactive Streams 契約以及與single、tick、cycle等相近操作符的選型對比。閱讀完本文你將掌握如何用Source.repeat構(gòu)造無限數(shù)據(jù)流并學(xué)會用take、grouped等下游操作符將其截斷為有限流從而安全地用于輪詢、心跳、模擬數(shù)據(jù)注入等實戰(zhàn)場景。一、操作符概述Source.repeat是 Akka Streams 提供的無限數(shù)據(jù)源構(gòu)造操作符它接收一個單一元素element然后反復(fù)發(fā)射同一個值。只要下游存在需求demand它就會源源不斷地向外發(fā)射該值并且永遠(yuǎn)不會自行完成complete。正因如此文檔明確提醒如果想讓這個流變成有限流必須與其它操作符如take、grouped、zip等組合使用在下游主動截斷它。在 Akka Streams 的官方操作符分類中Source.repeat屬于 Source operators 大家族是構(gòu)建數(shù)據(jù)源的最基礎(chǔ)操作符之一。二、方法簽名Source.repeat同時提供 Scala 與 Java 兩個 API 入口簽名如下Scala定義于 akka-stream/src/main/scala/akka/stream/scaladsl/Source.scaladef repeatT: Source[T, NotUsed]Java定義于 akka-stream/src/main/scala/akka/stream/javadsl/Source.scalapublic static T SourceT, NotUsed repeat(T element)簽名要點(diǎn)返回類型Source[T, NotUsed]NotUsed作為 materialized value 類型表示該數(shù)據(jù)源在被物化時不產(chǎn)生任何有價值的運(yùn)行期對象與tick返回Source[T, Cancellable]形成對比后者物化后可用來取消定時器。泛型參數(shù)T被重復(fù)發(fā)射的元素類型完全由傳入值推斷可以是任意類型——整數(shù)、字符串、消息對象、配置項皆可。三、Reactive Streams 語義依據(jù)官方文檔的div { .callout }定義Source.repeat的 Reactive Streams 契約如下行為描述emits發(fā)射當(dāng)下游存在需求時反復(fù)發(fā)射同一個值completes完成永遠(yuǎn)不會自行完成這兩條語義直接決定了它的使用邊界由于“永遠(yuǎn)不完成”直接對Source.repeat調(diào)用runForeach或接入Sink.foreach而不做任何截斷程序?qū)o限運(yùn)行下去。由于“按需發(fā)射”它天然遵守背壓backpressure協(xié)議——下游不需要時它不會繼續(xù)發(fā)射下游消費(fèi)多快它就生產(chǎn)多快不會造成緩沖區(qū)無限膨脹。四、源碼級實現(xiàn)原理4.1 Scala 側(cè)實現(xiàn)Source.repeat的 Scala 實現(xiàn)非常精簡scaladsl/Source.scala#L428-L430def repeatT: Source[T, NotUsed] { fromIterator(() Iterator.continually(element)).withAttributes(DefaultAttributes.repeat) }實現(xiàn)拆解Iterator.continually(element)是 Scala 標(biāo)準(zhǔn)庫提供的一個無限迭代器每次調(diào)用next()都返回同一個element永不耗盡。fromIterator(() ...)使用按需求值的函數(shù)包裝迭代器意味著迭代器是惰性創(chuàng)建的——只有在該 Source 真正被物化materialize且下游開始請求元素時才會創(chuàng)建而不是在調(diào)用repeat的瞬間就生成。.withAttributes(DefaultAttributes.repeat)為這個階段打上名為repeat的屬性標(biāo)簽。該屬性定義于 akka-stream/src/main/scala/akka/stream/impl/Stages.scala#L95val repeat name(repeat)用于流調(diào)試debug 階段名、監(jiān)控和日志輸出時標(biāo)識該階段。從源碼結(jié)構(gòu)可以推斷Source.repeat底層本質(zhì)上是一個無限迭代器數(shù)據(jù)源與Source.fromIterator、Source.cycle共享同一套迭代器基礎(chǔ)設(shè)施Stages.scala#L105 中的cycledSource即用于cycle的同類屬性。它的內(nèi)存占用與元素個數(shù)無關(guān)——無論發(fā)射 1 次還是 10 億次都只保存這一個元素本身。4.2 Java 側(cè)實現(xiàn)Java API 是 Scala 實現(xiàn)的一層薄封裝javadsl/Source.scala#L245-L246def repeatT: Source[T, NotUsed] new Source(scaladsl.Source.repeat(element))Java 用戶調(diào)用Source.repeat(element)時內(nèi)部委托給 Scala 的scaladsl.Source.repeat并包裝成akka.stream.javadsl.Source從而保證 Java 與 Scala 在行為、性能上完全一致。五、實戰(zhàn)示例5.1 Scala 示例官方文檔的 Scala 示例取自測試文件 akka-stream-tests/src/test/scala/akka/stream/scaladsl/SourceSpec.scala#L260-L270演示了如何用take(4)截斷無限流只打印前 4 個元素val source: Source[Int, NotUsed] Source.repeat(42) val f source.take(4).runWith(Sink.foreach(println)) // 輸出 // 42 // 42 // 42 // 425.2 Java 示例對應(yīng)的 Java 版本取自 akka-stream-tests/src/test/java/akka/stream/javadsl/SourceTest.java#L638-L650SourceInteger, NotUsed source Source.repeat(42); CompletionStageDone f source.take(4).runWith(Sink.foreach(System.out::println), system); // 輸出 // 42 // 42 // 42 // 425.3 批量消費(fèi)示例驗證無限發(fā)射測試套件中還有兩個更有說服力的用例證明repeat確實“無限”且“同值”ScalaSourceSpec.scala#L253-L258repeat as long as it takes in { val f Source.repeat(42).grouped(1000).runWith(Sink.head) f.futureValue.size should (1000) f.futureValue.toSet should (Set(42)) }JavaSourceTest.java#L630-L636final CompletionStageListInteger f Source.repeat(42).grouped(10000).runWith(Sink.head(), system); final ListInteger result f.toCompletableFuture().get(3, TimeUnit.SECONDS); assertEquals(10000, result.size()); for (Integer i : result) assertEquals(i, (Integer) 42);兩個用例都驗證了repeat可以一口氣吐出 1000 乃至 10000 個元素且全部等于同一值而grouped(n).runWith(Sink.head)則巧妙地把無限流“采樣”成有限結(jié)果——這也是一種常用的無限流截斷技巧。5.4 物化值類型NotUsed的意義repeat返回的Source[T, NotUsed]在物化后不產(chǎn)生可操作的句柄因此它適合作為純數(shù)據(jù)發(fā)生器接入圖Graph或與其它 Source 合并。如果你需要能在運(yùn)行期停止/取消的周期發(fā)生器應(yīng)改用tick。六、與相近操作符的對比與選型官方文檔在repeat頁面末尾專門列出了三個“參見”操作符它們共同構(gòu)成“重復(fù)發(fā)射類”數(shù)據(jù)源的完整譜系操作符行為完成時機(jī)物化值適用場景single只發(fā)射一次單個對象發(fā)射后立即完成NotUsed一次性消息、初始化數(shù)據(jù)repeat反復(fù)發(fā)射同一個對象永不完成NotUsed模擬數(shù)據(jù)注入、無限常量流tick按固定時間間隔周期發(fā)射永不完成可取消Cancellable心跳、輪詢、定時采樣cycle循環(huán)遍歷一個迭代器永不完成空迭代器拋異常NotUsed循環(huán)播放序列、輪流分發(fā)選型要點(diǎn)只需一次發(fā)射 → 用single。需要相同值持續(xù)發(fā)射、且不關(guān)心節(jié)奏 → 用repeat。需要按時間節(jié)奏發(fā)射即使相同→ 用tick因為tick自帶initialDelay與interval且物化出的Cancellable可隨時取消。需要循環(huán)一組不同的值如List(1, 2, 3)循環(huán)→ 用cycle傳入迭代器工廠。值得注意的差異點(diǎn)repeat和cycle都基于迭代器實現(xiàn)但repeat固定返回同一個對象引用而cycle每次迭代遍歷的是同一個迭代器的循環(huán)——當(dāng)原始迭代器耗盡后會重新從頭部開始且若傳入的是空迭代器會以異常終止流見 cycle.md 的說明。此外repeat是無節(jié)奏的“盡力而為”發(fā)射tick則是嚴(yán)格定時發(fā)射二者在需要限速的場景不可互換。七、常見用法與注意事項7.1 必須截斷無限流的三大經(jīng)典截斷手段由于repeat永不完成實戰(zhàn)中幾乎總是與以下操作符之一組合take(n)只取前 n 個元素如示例中的take(4)適合“固定數(shù)量”場景grouped(n).runWith(Sink.head)先按 n 個一批聚合再取第一批適合“批量取樣”場景見測試用例zip與有限流例如Source.repeat(template).zip(Source(1 to 100))讓有限流“耗盡”時帶動無限流結(jié)束。7.2 背壓與資源安全repeat遵循 Reactive Streams 背壓協(xié)議不會主動向內(nèi)存中堆積元素配合take、grouped等操作符后下游停止消費(fèi)時上游自然暫停。這使它成為在測試中注入海量同構(gòu)數(shù)據(jù)的首選——例如壓力測試中連續(xù)喂入相同請求消息或者為演示程序生成無限的同值序列。7.3 引用語義提示repeat反復(fù)發(fā)射的是同一個對象引用。若元素是可變對象下游多個消費(fèi)者會共享同一實例可能引發(fā)并發(fā)修改問題需要每次發(fā)射獨(dú)立副本時請配合map進(jìn)行拷貝或改用cycle 迭代器工廠、unfold等按需生成新實例的操作符。八、小結(jié)Source.repeat(element)構(gòu)造一個按背壓反復(fù)發(fā)射同一元素、永不完成的無限數(shù)據(jù)源簽名見 scaladsl/Source.scalaJava 封裝見 javadsl/Source.scala。其底層實現(xiàn)為fromIterator(() Iterator.continually(element))惰性創(chuàng)建、內(nèi)存恒定階段屬性repeat定義于 impl/Stages.scala#L95。完整可運(yùn)行示例與行為驗證可分別在 SourceSpec.scala 與 SourceTest.java 中找到。選型口訣一次用single同值無限用repeat定時無限用tick循環(huán)序列用cycle凡是使用repeat務(wù)必記住“永不完成”在下游用take、grouped或與有限流zip來主動收束流。贊分享后端并發(fā)編程異步編程【免費(fèi)下載鏈接】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點(diǎn)擊查看免費(fèi)下載相關(guān)推薦Akka Streams initialDelay 操作符完全指南源碼實現(xiàn)與實戰(zhàn)用法Akka Streams initialDelay 操作符完全指南源碼實現(xiàn)與實戰(zhàn)用法 導(dǎo)讀 initialDelay 是 Akka Streams 中一個輕量后端并發(fā)編程異步編程Akka Streams prependLazy 操作符詳解惰性前插 Source 的實現(xiàn)原理與實戰(zhàn)用法Akka Streams prependLazy 操作符詳解惰性前插 Source 的實現(xiàn)原理與實戰(zhàn)用法 prependLazy 是 Akka Streams后端并發(fā)編程異步編程Akka PubSub.source 操作符全解將 Typed Topic 訂閱接入 Akka Streams 數(shù)據(jù)流Akka PubSub.source 操作符全解將 Typed Topic 訂閱接入 Akka Streams 數(shù)據(jù)流 PubSub.source 是 akk后端并發(fā)編程異步編程創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考