后端并发编程异步编程【免费下载链接】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点击查看免费下载Source.lazilyAsync是 Akka Streams 中一个典型的按需惰性lazy类算子它把Future/CompletionStage的创建和物化推迟到下游产生需求demand的那一刻从而避免昂贵的异步资源在流刚启动时就被白白创建。该算子自 Akka 2.6.0 起被标记为废弃deprecated官方推荐迁移到lazyFutureSource或更细粒度的lazyFuture/lazySource。本文以 akka-docs/src/main/paradox/stream/operators/Source/lazilyAsync.md 为骨架结合 akka-stream 模块的源码与测试完整讲解其行为语义、内部实现、边界场景并给出可直接落地的迁移代码。一、算子定位lazilyAsync 是什么lazilyAsync属于 Akka Streams 的 Source 算子族。它的核心行为可以用一句话概括延迟一个CompletionStage的创建与物化直到下游出现需求demand为止。也就是说你提供给算子的不是已经创建好的Future而是一个将来才会执行的工厂函数factory。只有当下游真正开始请求元素时工厂才会被调用、Future才会被创建并参与流的物化。值得注意的是当前版本中该文档明确标注Deprecated bySource.lazyFutureSource。原文档见 lazilyAsync.mdlazilyAsynchas been deprecated in 2.6.0, uselazyFutureSourceinstead。因此阅读本文时重点是理解惰性 Future 源这一类算子的设计意图与语义而不是在新代码中继续使用lazilyAsync。二、签名Scala 与 Java 双 DSLScala DSLdef lazilyAsyncT Future[T]): Source[T, Future[NotUsed]]对应实现在 akka-stream/src/main/scala/akka/stream/scaladsl/Source.scala#L509-L511deprecated(Use Source.lazyFuture instead, 2.6.0) def lazilyAsyncT Future[T]): Source[T, Future[NotUsed]] lazily(() fromFuture(create()))注意两个细节参数是按名传值的工厂函数() Future[T]而不是直接的Future[T]——这正是惰性的来源物化值类型是Future[NotUsed]即内部Future被创建并物化这一事件本身是以一个异步Future暴露给外部的。Java DSLpublic T SourceT, FutureNotUsed lazilyAsync(CreatorCompletionStageT create)对应实现在 akka-stream/src/main/scala/akka/stream/javadsl/Source.scala#L283-L285deprecated(Use Source.lazyCompletionStage instead, 2.6.0) public T SourceT, FutureNotUsed lazilyAsync(CreatorCompletionStageT create) { return scaladsl.Source.lazilyAsync(() - create.create().asScala).asJava(); }Java 侧使用akka.japi.function.CreatorCompletionStageT与 Scala 的() Future[T]一一对应。三、行为语义Reactive Streams 视角原文档给出了该算子在 Reactive Streams 语义层面的承诺见 lazilyAsync.md语义说明emits发射当内部Future完成时发射其值completes完成在内部Future完成后流完成展开来讲流在没有下游需求之前不会调用工厂函数、不会创建Future一旦下游发起首次请求pull工厂被调用返回的Future被接入流中Future成功完成后其值作为单个元素发射给下游随后流正常完成如果Future以失败结束流会以该异常失败fail异常会传递给下游订阅者。这套语义与文档中emits/completes两条承诺完全吻合内部Future本质上等价于一个只发射单元素的异步源。四、源码级原理一次pull引发的连锁物化lazilyAsync的实现非常短——它只是两层现成算子的组合lazily(() fromFuture(create()))4.1 外层lazily现为lazySourcelazily本身也已在 2.6.0 被废弃deprecated(Use Source.lazySource instead, 2.6.0)见 Source.scala#L498-L500它的实现委托给内部的LazySource图阶段GraphStagedeprecated(Use Source.lazySource instead, 2.6.0) def lazilyT, M Source[T, M]): Source[T, Future[M]] Source.fromGraph(new LazySourceT, M)4.2 内层fromFuturefromFuture把单个Future[T]包装成一个只发射一个元素的Source同样已废弃建议用Source.future见 Source.scala#L374-L376deprecated(Use Source.future instead, 2.6.0) def fromFutureT: Source[T, NotUsed] fromGraph(new FutureSource(future))4.3 真正干活的LazySource图阶段外层LazySource是理解惰性的关键它位于 akka-stream/src/main/scala/akka/stream/impl/LazySource.scala标注为InternalApi即内部 API。其核心逻辑在onPull()回调中LazySource.scala#L43-L80override def onPull(): Unit { val source try { sourceFactory() // 1. 此刻才调用工厂 } catch { case NonFatal(ex) matPromise.tryFailure(ex) throw ex } val subSink new SubSinkInletT subSink.pull() // ... 把内部 source 物化到 subSink并把物化值写入 matPromise try { val matVal subFusingMaterializer.materialize(source.toMat(subSink.sink)(Keep.left), inheritedAttributes) matPromise.trySuccess(matVal) } catch { case NonFatal(ex) subSink.cancel() failStage(ex) matPromise.tryFailure(ex) } }整个生命周期可以拆解为等待需求LazySource被物化后仅仅是挂着一个matPromise类型Promise[M]等待不会做任何额外工作首次onPull()下游第一次 pull 时才真正调用sourceFactory()若工厂抛出异常matPromise立即失败tryFailure(ex)异常沿流传播嵌套物化工厂返回的Source通过subFusingMaterializer物化并与SubSinkInlet对接实现外层源切换到内层源物化成功后matPromise以内部源的物化值完成trySuccess(matVal)物化失败则failStage(ex)并让matPromise失败切换后的透传切换完成后out的onPull直接转发为subSink.pull()内层元素通过onPush推给下游LazySource.scala#L54-L69。两个值得注意的失败路径源码注释与实现一致下游在无需求时取消onDownstreamFinish中matPromise.failure(new NeverMaterializedException(cause))LazySource.scala#L38-L41即物化值携带NeverMaterializedException阶段异常终止postStop中若matPromise尚未完成则用AbruptStageTerminationException失败LazySource.scala#L84-L86。五、测试佐证行为边界由测试锁定仓库中 akka-stream-tests/src/test/scala/akka/stream/scaladsl/LazilyAsyncSpec.scala 用一组测试精确锁定了lazilyAsync的行为测试场景验证结论work in happy path scenarioSource.lazilyAsync(() Future(42)).runWith(Sink.head)得到42call factory method on demand only用AtomicBoolean标记工厂调用下游probe.cancel()后constructed.get() false证明无需求则不调用工厂fail materialized value when downstream cancels without ever consuming any element下游Sink.cancelled时物化Future以RuntimeExceptionNeverMaterializedException失败materialize when the source has been created工厂被调用后物化Future才完成matF.value None直到内部源物化propagate failed future from factory工厂返回Future.failed(failure)时流以该failure失败这些测试直接对应原文档Defers creation and materialization of aCompletionStageuntil there is demand的行为描述也印证了第四节源码中的各条失败路径。另外 LazySourceSpec.scala 还覆盖了lazySource/lazyFutureSource的同类边界内部源物化失败、工厂抛异常、下游取消后停止消费、以及内部源可继承外部Attributes等。六、为什么被废弃2.6.0 起的算子重组lazilyAsync在2.6.0被废弃。废弃的核心原因是 Akka Streams 在 2.6.0 对惰性系列算子做了一次系统化重组把原来混在一起的三种惰性能力拆分成职责单一的算子并统一了命名lazy*前缀。6.1 替代算子对照表原算子废弃说明源码注解推荐替代lazilyAsyncScalaUse Source.lazyFuture insteadSource.scala#L509lazyFuture延迟单个元素的 FuturelazilyAsyncJavaUse Source.lazyCompletionStage insteadSource.scala#L283lazyCompletionStagelazilyAsync算子文档use lazyFutureSource insteadlazilyAsync.mdlazyFutureSource延迟Future[Source]lazilyScalaUse Source.lazySource insteadlazySource延迟SourcefromFutureUse Source.future insteadfuturefromFutureSourceUse Source.futureSource配合fromGraphfutureSource说明算子文档指向lazyFutureSource而源码中的deprecated注解指向lazyFuture/lazyCompletionStage。二者并不矛盾——lazilyAsync内部是惰性 单元素 Future最直接的等价替代是lazyFuture如果你需要的是惰性 Future[Source] 的完整源切换则对应更通用的lazyFutureSource。迁移时按实际需求选择即可。6.2 现代替代算子的签名与实现新的惰性算子同样位于 akka-stream/src/main/scala/akka/stream/scaladsl/Source.scalalazyFuture——延迟单元素 FutureSource.scala#L573-L574def lazyFutureT Future[T]): Source[T, NotUsed] single(()).mapAsyncUnordered(1)(_ create())lazySource——延迟完整 SourceSource.scala#L592-L593def lazySourceT, M Source[T, M]): Source[T, Future[M]] fromGraph(new LazySource(create))lazyFutureSource——延迟 Future[Source]即lazilyAsync文档推荐的目标Source.scala#L612-L613def lazyFutureSourceT, M Future[Source[T, M]]): Source[T, Future[M]] lazySource(() futureSource(create())).mapMaterializedValue(_.flatten)futureSource——把已完成的Future[Source]拍平为源Source.scala#L545-L551def futureSourceT, M: Source[T, Future[M]] { futureSource.value match { case Some(Success(source)) source.mapMaterializedValue(Future.successful) case Some(Failure(exc)) failed(exc).mapMaterializedValue(_ Future.failed(exc)) case _ fromGraph(new FutureFlattenSource(futureSource)) } }Java DSL 对应提供lazyCompletionStage、lazySource、lazyCompletionStageSource见 javadsl/Source.scala#L350-L393其中lazyCompletionStageSource在内部复用completionStageSource并做thenCompose拍平。6.3 迁移示例场景 A延迟一个单元素 FuturelazilyAsync→lazyFutureScala// 废弃写法 val oldSource: Source[Int, Future[NotUsed]] Source.lazilyAsync(() Future { fetchExpensiveValue() }) // 推荐写法 val newSource: Source[Int, NotUsed] Source.lazyFuture(() Future { fetchExpensiveValue() })Java// 废弃写法 SourceInteger, FutureNotUsed oldSource Source.lazilyAsync(() - CompletableFuture.supplyAsync(() - fetchExpensiveValue())); // 推荐写法 SourceInteger, NotUsed newSource Source.lazyCompletionStage(() - CompletableFuture.supplyAsync(() - fetchExpensiveValue()));场景 B延迟一个 Future[Source]原文档推荐方向Source.lazyFutureSource { () Future { // 昂贵的源创建逻辑仅在首次需求时执行 createExpensiveSource() } }七、实战注意事项惰性的边界与反直觉行为7.1 预取会破坏惰性lazilyAsync及整个lazy*家族的文档都反复强调同一句警告见 lazyFutureSource.md 与 lazySource.md流中的异步边界asynchronous boundaries和其他算子可能进行预取pre-fetching这会抵消惰性导致工厂被立即触发。也就是说惰性是尽力而为的只要整条流上没有异步边界Sink.head、Sink.seq这类按需拉取的 sink 能保证工厂在下游第一次 pull 时才执行但一旦链路中出现async、buffer、Sink.queue等带缓冲的算子缓冲区的预取请求会立刻转化为对工厂的调用。lazySource.md文档中的反例akka-docs/src/test/scala/docs/stream/operators/source/Lazy.scala#L18-L32演示了这一点val source Source.lazySource { () println(Creating the actual source) createExpensiveSource() } val queue source.runWith(Sink.queue()) // ... 时间流逝 ... // 你以为第一次 pull 才创建源 // 但 Sink.queue 在物化时就会缓冲并立即请求元素 // 因此上面的 println 早已打印源已被创建 queue.pull()7.2 真正的价值每个物化各有一份工厂产物lazySource.md同时指出该算子最有用的特性是工厂每次物化只调用一次。因此可以用它安全地为每次运行materialization构造独立的可变对象避免同一个可变实例在多次run()之间被不安全地共享见 Lazy.scala#L38-L55 的IteratorLikeThing示例。这个结论对lazilyAsync家族同样成立。7.3 物化值的失败语义不要忘记lazilyAsync的物化值本身是一个Future。它的完成/失败时机由 LazilyAsyncSpec.scala 锁定为工厂被调用且内部源成功物化 → 物化Future成功工厂抛异常 / 内部Future失败 / 内部源物化失败 → 物化Future失败下游在无需求时取消或失败 → 物化Future以NeverMaterializedException失败。如果你的业务需要知道惰性源是否真的被创建了请通过mapMaterializedValue观察并处理这个Future。八、总结Source.lazilyAsync虽然已在 2.6.0 废弃但它所代表的延迟异步资源的创建与物化直到下游需求出现这一设计模式至今仍是 Akka Streams 惰性算子家族的核心。理解它需要抓住三条主线行为无需求不创建Future需求到达后才创建、物化并发射单元素然后完成对应 Reactive Streams 语义中的emits/completes实现lazilyAsync lazily(() fromFuture(create()))底层由LazySource图阶段在首次onPull时触发嵌套物化物化事件通过Promise暴露给外部迁移新代码按需求选择lazyFuture单元素、lazySource完整源或lazyFutureSourceFuture[Source]Java 侧对应lazyCompletionStage/lazySource/lazyCompletionStageSource并始终警惕异步边界与预取对惰性的破坏。相关资源算子文档lazilyAsync.md、lazyFutureSource.md、lazySource.md、lazyFuture.md、Source 算子索引核心实现scaladsl/Source.scala、javadsl/Source.scala、impl/LazySource.scala测试佐证LazilyAsyncSpec.scala、LazySourceSpec.scala示例代码docs/stream/operators/source/Lazy.scala赞分享后端并发编程异步编程【免费下载链接】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 drop 算子丢弃前 N 个元素的操作语义与源码实现解析Akka Streams drop 算子丢弃前 N 个元素的操作语义与源码实现解析 导读 drop 是 Akka Streams 中用于丢弃流中前 n 个元后端并发编程异步编程Akka Streams filter 操作符全解析谓词过滤、源码实现与 Reactive Streams 语义Akka Streams filter 操作符全解析谓词过滤、源码实现与 Reactive Streams 语义 本文是一份面向 Akka Streams 开后端并发编程异步编程Akka Streams map 操作符完全指南逐元素变换、Reactive Streams 语义与源码实现解析Akka Streams map 操作符完全指南逐元素变换、Reactive Streams 语义与源码实现解析 map 是 Akka Streams 中最基后端并发编程异步编程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考