后端并发编程异步编程【免费下载链接】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 官方文档 akka-docs/src/main/paradox/stream/operators/Source/repeat.md 为主体结合 akka-stream 模块的源码与测试用例系统讲解Source.repeat操作符的签名、语义、底层实现、Reactive Streams 契约以及与single、tick、cycle等相近操作符的选型对比。阅读完本文你将掌握如何用Source.repeat构造无限数据流并学会用take、grouped等下游操作符将其截断为有限流从而安全地用于轮询、心跳、模拟数据注入等实战场景。一、操作符概述Source.repeat是 Akka Streams 提供的无限数据源构造操作符它接收一个单一元素element然后反复发射同一个值。只要下游存在需求demand它就会源源不断地向外发射该值并且永远不会自行完成complete。正因如此文档明确提醒如果想让这个流变成有限流必须与其它操作符如take、grouped、zip等组合使用在下游主动截断它。在 Akka Streams 的官方操作符分类中Source.repeat属于 Source operators 大家族是构建数据源的最基础操作符之一。二、方法签名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)签名要点返回类型Source[T, NotUsed]NotUsed作为 materialized value 类型表示该数据源在被物化时不产生任何有价值的运行期对象与tick返回Source[T, Cancellable]形成对比后者物化后可用来取消定时器。泛型参数T被重复发射的元素类型完全由传入值推断可以是任意类型——整数、字符串、消息对象、配置项皆可。三、Reactive Streams 语义依据官方文档的div { .callout }定义Source.repeat的 Reactive Streams 契约如下行为描述emits发射当下游存在需求时反复发射同一个值completes完成永远不会自行完成这两条语义直接决定了它的使用边界由于“永远不完成”直接对Source.repeat调用runForeach或接入Sink.foreach而不做任何截断程序将无限运行下去。由于“按需发射”它天然遵守背压backpressure协议——下游不需要时它不会继续发射下游消费多快它就生产多快不会造成缓冲区无限膨胀。四、源码级实现原理4.1 Scala 侧实现Source.repeat的 Scala 实现非常精简scaladsl/Source.scala#L428-L430def repeatT: Source[T, NotUsed] { fromIterator(() Iterator.continually(element)).withAttributes(DefaultAttributes.repeat) }实现拆解Iterator.continually(element)是 Scala 标准库提供的一个无限迭代器每次调用next()都返回同一个element永不耗尽。fromIterator(() ...)使用按需求值的函数包装迭代器意味着迭代器是惰性创建的——只有在该 Source 真正被物化materialize且下游开始请求元素时才会创建而不是在调用repeat的瞬间就生成。.withAttributes(DefaultAttributes.repeat)为这个阶段打上名为repeat的属性标签。该属性定义于 akka-stream/src/main/scala/akka/stream/impl/Stages.scala#L95val repeat name(repeat)用于流调试debug 阶段名、监控和日志输出时标识该阶段。从源码结构可以推断Source.repeat底层本质上是一个无限迭代器数据源与Source.fromIterator、Source.cycle共享同一套迭代器基础设施Stages.scala#L105 中的cycledSource即用于cycle的同类属性。它的内存占用与元素个数无关——无论发射 1 次还是 10 亿次都只保存这一个元素本身。4.2 Java 侧实现Java API 是 Scala 实现的一层薄封装javadsl/Source.scala#L245-L246def repeatT: Source[T, NotUsed] new Source(scaladsl.Source.repeat(element))Java 用户调用Source.repeat(element)时内部委托给 Scala 的scaladsl.Source.repeat并包装成akka.stream.javadsl.Source从而保证 Java 与 Scala 在行为、性能上完全一致。五、实战示例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 示例对应的 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 批量消费示例验证无限发射测试套件中还有两个更有说服力的用例证明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)则巧妙地把无限流“采样”成有限结果——这也是一种常用的无限流截断技巧。5.4 物化值类型NotUsed的意义repeat返回的Source[T, NotUsed]在物化后不产生可操作的句柄因此它适合作为纯数据发生器接入图Graph或与其它 Source 合并。如果你需要能在运行期停止/取消的周期发生器应改用tick。六、与相近操作符的对比与选型官方文档在repeat页面末尾专门列出了三个“参见”操作符它们共同构成“重复发射类”数据源的完整谱系操作符行为完成时机物化值适用场景single只发射一次单个对象发射后立即完成NotUsed一次性消息、初始化数据repeat反复发射同一个对象永不完成NotUsed模拟数据注入、无限常量流tick按固定时间间隔周期发射永不完成可取消Cancellable心跳、轮询、定时采样cycle循环遍历一个迭代器永不完成空迭代器抛异常NotUsed循环播放序列、轮流分发选型要点只需一次发射 → 用single。需要相同值持续发射、且不关心节奏 → 用repeat。需要按时间节奏发射即使相同→ 用tick因为tick自带initialDelay与interval且物化出的Cancellable可随时取消。需要循环一组不同的值如List(1, 2, 3)循环→ 用cycle传入迭代器工厂。值得注意的差异点repeat和cycle都基于迭代器实现但repeat固定返回同一个对象引用而cycle每次迭代遍历的是同一个迭代器的循环——当原始迭代器耗尽后会重新从头部开始且若传入的是空迭代器会以异常终止流见 cycle.md 的说明。此外repeat是无节奏的“尽力而为”发射tick则是严格定时发射二者在需要限速的场景不可互换。七、常见用法与注意事项7.1 必须截断无限流的三大经典截断手段由于repeat永不完成实战中几乎总是与以下操作符之一组合take(n)只取前 n 个元素如示例中的take(4)适合“固定数量”场景grouped(n).runWith(Sink.head)先按 n 个一批聚合再取第一批适合“批量取样”场景见测试用例zip与有限流例如Source.repeat(template).zip(Source(1 to 100))让有限流“耗尽”时带动无限流结束。7.2 背压与资源安全repeat遵循 Reactive Streams 背压协议不会主动向内存中堆积元素配合take、grouped等操作符后下游停止消费时上游自然暂停。这使它成为在测试中注入海量同构数据的首选——例如压力测试中连续喂入相同请求消息或者为演示程序生成无限的同值序列。7.3 引用语义提示repeat反复发射的是同一个对象引用。若元素是可变对象下游多个消费者会共享同一实例可能引发并发修改问题需要每次发射独立副本时请配合map进行拷贝或改用cycle 迭代器工厂、unfold等按需生成新实例的操作符。八、小结Source.repeat(element)构造一个按背压反复发射同一元素、永不完成的无限数据源签名见 scaladsl/Source.scalaJava 封装见 javadsl/Source.scala。其底层实现为fromIterator(() Iterator.continually(element))惰性创建、内存恒定阶段属性repeat定义于 impl/Stages.scala#L95。完整可运行示例与行为验证可分别在 SourceSpec.scala 与 SourceTest.java 中找到。选型口诀一次用single同值无限用repeat定时无限用tick循环序列用cycle凡是使用repeat务必记住“永不完成”在下游用take、grouped或与有限流zip来主动收束流。赞分享后端并发编程异步编程【免费下载链接】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 initialDelay 操作符完全指南源码实现与实战用法Akka Streams initialDelay 操作符完全指南源码实现与实战用法 导读 initialDelay 是 Akka Streams 中一个轻量后端并发编程异步编程Akka Streams prependLazy 操作符详解惰性前插 Source 的实现原理与实战用法Akka Streams prependLazy 操作符详解惰性前插 Source 的实现原理与实战用法 prependLazy 是 Akka Streams后端并发编程异步编程Akka PubSub.source 操作符全解将 Typed Topic 订阅接入 Akka Streams 数据流Akka PubSub.source 操作符全解将 Typed Topic 订阅接入 Akka Streams 数据流 PubSub.source 是 akk后端并发编程异步编程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考