后端并发编程异步编程【免费下载链接】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点击查看免费下载导读mapAsyncUnordered是 Akka Streams 中用于异步并发处理的核心操作符之一。它与mapAsync类似会将每个上游元素映射为一个FutureScala/CompletionStageJava但不保证下游输出顺序——哪个异步任务先完成结果就先被发射。本文基于当前仓库的官方文档与源码实现完整讲解其签名、语义、Reactive Streams 行为、与mapAsync的取舍并结合 Ops.scala 中的底层实现剖析其工作原理帮助读者在消息处理、HTTP 调用等对吞吐量敏感、对顺序无要求的场景中正确使用该操作符。操作符签名mapAsyncUnordered同时定义在Source与Flow上Scala 与 Java 的签名分别如下ScalaFlow.scaladef mapAsyncUnorderedT(f: Out Future[T]): Repr[T]JavamapAsyncUnordered(int parallelism, akka.japi.function.FunctionOut, CompletionStageT)其中parallelism最大并行度即同一时刻最多有多少个由f返回的Future/CompletionStage在途in-flight。底层实现中该值同时决定内部缓冲区的容量见下文实现原理。f作用于每个上游元素、返回异步结果的映射函数。核心语义与 mapAsync 的差异结果按完成顺序发射mapAsyncUnordered与mapAsync一样都是异步映射操作符二者的区别仅在输出顺序上mapAsync保证输出顺序与输入顺序一致先提交的任务结果必须先被发射即使它后完成也要阻塞等待mapAsyncUnordered结果按完成顺序发射哪个Future/CompletionStage先完成就先传递哪个与触发它的元素在流中的先后位置无关。因此当元素之间互不相关、顺序没有业务意义时例如一批相互独立的消息、一次批量的外部 API 调用mapAsyncUnordered可以避免mapAsync因队头阻塞head-of-line blocking造成的吞吐量损失。null 结果与失败处理如果某个Future/CompletionStage以null完成该结果会被忽略直接继续处理下一个元素不向下游发射任何值。如果某个Future/CompletionStage失败流默认也会失败fail除非为该操作符配置了其他监督策略supervision strategy例如Supervision.resumingDecider跳过失败元素、Supervision.restartingDecider重启后继续。关于监督策略的详细配置方式可参考 stream-error.md 文档。实战示例乱序并发处理消息流官方文档给出了一个非常贴近真实场景的示例从某个 Source 消费消息由于元素彼此不相关顺序无关优先追求吞吐量而非顺序于是使用mapAsyncUnordered配合一定并行度让消息处理完即发射。Scala 示例完整可运行代码位于 MapAsyncs.scalaimport CommonMapAsync._ events .mapAsyncUnordered(3) { in eventHandler(in) } .map { in println(smapAsyncUnordered emitted event number: $in) } .runWith(Sink.ignore)其中events是一个经过throttle(1, 50.millis)限速的事件流每 50ms 一个元素而eventHandler模拟了一个耗时不定的异步处理过程——约 1/5 的概率延迟 500ms 才完成其余立即完成def eventHandler(event: Event): Future[Int] { println(sProcessing event $event...) val result if (Random.nextInt(5) 0) { akka.pattern.after(500.millis)(Future.successful(event.sequenceNumber)) } else { Future.successful(event.sequenceNumber) } result.map { x println(sCompleted processing $x) x } }Java 示例对应 Java 版本完整代码位于 MapAsyncs.javaevents .mapAsyncUnordered(10, this::eventHandler) .map(in - mapSync emitted event number in.intValue()) .runWith(Sink.foreach(str - System.out.println(str)), system);Java 中eventHandler返回CompletionStageInteger并使用Patterns.after模拟约 1/5 概率的 500ms 延迟。运行日志解读运行上述流时日志输出大致如下文档原样摘录[...] Processing event numner Event(27)... Completed processing 27 mapAsyncUnordered emitted event number: 27 Processing event numner Event(28)... Completed processing 22 mapAsyncUnordered emitted event number: 22 Processing event numner Event(29)... Completed processing 26 mapAsyncUnordered emitted event number: 26 Processing event numner Event(30)... Completed processing 30 mapAsyncUnordered emitted event number: 30 Processing event numner Event(31)... Completed processing 31 mapAsyncUnordered emitted event number: 31 [...]可以清楚看到元素22 的处理完成先于 28因此 22 先于 28 被发射26 也先于 29 被发射。这正是乱序发射的表现——发射顺序由异步任务的完成时间决定而不是由元素进入流的顺序决定。与之形成对比的是mapAsync会严格按输入顺序发射参见 mapAsync 文档 中的示例。Reactive Streams 语义依据官方文档mapAsyncUnordered在 Reactive Streams 协议下的行为契约如下语义说明emits发射只要函数返回的任意一个Future/CompletionStage完成其结果就会被发射backpressures背压当在途Future/CompletionStage的数量达到配置的parallelism且下游仍在背压时停止从上游拉取元素completes完成当上游完成、所有Future/CompletionStage均已完成、且所有元素都已发射时流完成底层实现原理源码视角mapAsyncUnordered的实际执行由内部GraphStage完成其实现位于 Ops.scala。理解这段实现有助于把握其性能特征与边界行为。三个核心状态private var inFlight 0 private var buffer: BufferImpl[Out] _ private val invokeFutureCB: Try[Out] Unit getAsyncCallback(futureCompleted).invokeinFlight当前在途未完成的Future数量buffer容量为parallelism的结果缓冲区在preStart中通过BufferImpl(parallelism, ...)初始化用于缓存已完成但下游尚未拉取的结果invokeFutureCB一个异步回调把Future的完成结果安全地送回流处理线程避免并发竞争。todo inFlight buffer.used表示尚未处理完的总量。元素处理与并行度控制在onPush中每收到一个上游元素就调用用户函数f并令inFlight 1override def onPush(): Unit { val future f(grab(in)) inFlight 1 future.value match { case None future.onComplete(invokeFutureCB)(ExecutionContext.parasitic) case Some(v) futureCompleted(v) } if (todo parallelism !hasBeenPulled(in)) tryPull(in) }这里有一个与mapAsync一致的优化若Future已经完成future.value非空则直接在当前线程调用futureCompleted处理结果而不必再经由调度器投递一次回调只有未完成的Future才注册onComplete回调并选用ExecutionContext.parasitic寄生执行上下文回调在原线程上直接执行减少调度开销。关键约束是if (todo parallelism !hasBeenPulled(in)) tryPull(in)——只有当inFlight buffer总量仍低于parallelism时才继续向上游拉取这正是背压语义中达到并行度即停止拉取的实现来源。结果发射与 null/失败分支futureCompleted处理每个Future的完成结果这是整个操作符的发射逻辑核心def futureCompleted(result: Try[Out]): Unit { def isCompleted isClosed(in) todo 0 inFlight - 1 result match { case Success(elem) if elem ! null if (isAvailable(out)) { if (!hasBeenPulled(in)) tryPull(in) push(out, elem) if (isCompleted) completeStage() } else buffer.enqueue(elem) case Success(_) if (isCompleted) completeStage() else if (!hasBeenPulled(in)) tryPull(in) case Failure(ex) if (decider(ex) Supervision.Stop) failStage(ex) else if (isCompleted) completeStage() else if (!hasBeenPulled(in)) tryPull(in) } }可以清晰对应到文档中描述的三种语义成功且非 null若下游此时正在拉取isAvailable(out)则立即push否则先存入缓冲区成功但为 null直接忽略该结果不发射、不入缓冲继续拉取下一个元素失败查询监督决策器decider若返回Supervision.Stop则failStage(ex)使整个流失败否则跳过该元素继续处理。此外onPull中会优先从缓冲区取出结果发射if (!buffer.isEmpty) push(out, buffer.dequeue())随后再根据todo与parallelism的关系决定是否补拉上游。整个阶段通过getAsyncCallback保证Future完成回调与流处理线程之间的线程安全。与 mapAsync 实现的关键差异对比同文件中的MapAsync实现Ops.scala可以发现mapAsync使用Holder包装结果并依赖缓冲区的FIFO 顺序——pushNextIfPossible中buffer.peek().elem eq NotYetThere时宁可等待也不越序发射ahead of line blocking to keep order而mapAsyncUnordered的buffer只缓存已完成的结果futureCompleted一旦完成即可尝试发射完全不受其他在途任务影响因此消除了顺序约束带来的队头阻塞。使用建议与注意事项适用场景元素间无顺序依赖、异步处理耗时差异大、追求最大吞吐的消息流/任务流如官方文档示例中的无关联消息消费。并行度选择parallelism决定同时在途的异步任务数需结合下游处理能力与资源连接池、线程池、数据库连接数等权衡它不是越大越好过大会导致下游背压频繁触发过小则并发收益不足。与 mapAsync 对比选型需要严格保序如按序提交、按序落库时使用mapAsync参考 mapAsync 文档顺序无关时优先mapAsyncUnordered。null 结果若Future以null完成结果会被静默忽略编写函数f时应注意这一点。失败与监督默认任一Future失败都会导致流失败可通过.withAttributes(SupervisionStrategy(...))或ActorAttributes.supervisionStrategy配置resuming/restarting策略实现跳过失败元素详见 stream-error.md。延伸阅读同类异步操作符mapAsync保序版见 mapAsync.mdmapAsyncPartitioned的分区保序实现可参考 MapAsyncs.scala异步操作符总览见 operators/index.md流错误处理与监督策略见 stream-error.md官方完整示例源码Scala 版、Java 版。赞分享后端并发编程异步编程【免费下载链接】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点击查看免费下载相关推荐CenterTaskbar三步实现Windows任务栏图标完美居中布局的终极方案CenterTaskbar三步实现Windows任务栏图标完美居中布局的终极方案 你是否厌倦了Windows任务栏图标默认左对齐的单调布局想要让桌面看起来更后端并发编程异步编程Akka Streams 异步算子完全指南mapAsync / mapAsyncUnordered / mapAsyncPartitioned 的并发、背压与顺序语义Akka Streams 异步算子完全指南mapAsync / mapAsyncUnordered / mapAsyncPartitioned 的并发、背压与后端并发编程异步编程EMQX消息吞吐量优化批处理与并发控制EMQX消息吞吐量优化批处理与并发控制 引言MQTT消息传输的性能瓶颈与解决方案 在物联网IoT和工业物联网IIoT场景中消息传输的吞吐量和可靠性后端物联网消息队列通信上一篇CodeGraph安装指南三平台一键部署与 Agent 快速接入下一篇react-admin Labeled 组件完全指南为 Field 组件添加标签的终极方案创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考