如何让Akka Stream Source Queue实现严格FIFO顺序执行？

你遇到的问题是Akka Streams默认的异步并行处理特性导致的——流会尽可能提前拉取元素并在各个阶段并行处理，所以会出现多个任务连续启动，然后按顺序完成的情况。要实现前一个任务完全完成后再启动下一个的严格串行执行，我们可以通过两种常见方式来调整流的处理逻辑：

方法一：使用mapAsync(1)限制并行度 #

mapAsync(1)会让流的这个阶段每次只处理一个元素，必须等当前元素的处理任务（Future）完成后，才会拉取下一个元素开始处理。这样就能强制实现串行执行：

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

val tradeQueue = Source.queue[TradeTask](1, OverflowStrategy.backpressure)
  // 用mapAsync(1)替代普通map，确保前一个任务启动并完成后再处理下一个
  .mapAsync(1) { task =>
    println("TradeTask Start："+task)
    // 如果你的ProcessA/C是异步操作，也需要在这里用Future包裹并确保串行
    Future {
      task // 这里可以直接传入后续Flow，或者在这里执行同步处理
    }
  }
  .via(ProcessA)
  .via(ProcessC)
  .via(ProcessC)
  .toMat(Sink.foreach(task => {
    log.info("TradeTask finish："+task)
  }))(Keep.left)
  .run()

// 可选：如果需要确保任务提交也是串行的，等待每个offer完成再提交下一个
import scala.concurrent.Await
import scala.concurrent.duration._

for (item <- 1 to 100) {
  val task = TradeTask(item)
  Await.result(tradeQueue.offer(task), 10.seconds)
}

注意点： #

如果你的ProcessA或ProcessC内部包含异步操作（比如调用了返回Future的方法），请确保这些Flow内部也使用mapAsync(1)来限制并行度，否则后续阶段依然可能出现并行处理的情况。

方法二：使用flatMapConcat将每个任务转为独立子流 #

flatMapConcat会把每个输入元素转换成一个独立的子流，并且严格按照顺序处理这些子流——只有前一个子流的所有元素都处理完毕，才会启动下一个子流的处理。这种方式更直观地保证了单个任务的完整生命周期串行：

val tradeQueue = Source.queue[TradeTask](1, OverflowStrategy.backpressure)
  .flatMapConcat { task =>
    // 为每个任务创建一个单元素子流，包含完整的处理链路
    Source.single(task)
      .map { t =>
        println("TradeTask Start："+t)
        t
      }
      .via(ProcessA)
      .via(ProcessC)
      .via(ProcessC)
  }
  .toMat(Sink.foreach(task => {
    log.info("TradeTask finish："+task)
  }))(Keep.left)
  .run()

这种方式下，每个TradeTask的启动、经过所有处理阶段、直到完成，都是一个完整的串行单元，完全不会和其他任务的处理过程重叠。

为什么原来的代码会乱序启动？ #

Akka Streams的设计是为了最大化吞吐量，默认会在各个阶段之间使用缓冲区，上游会尽可能提前推送元素到下游阶段，所以即使map是同步操作，流也会提前拉取多个元素，导致多个任务连续启动。而上面的两种方法都强制让流等待前一个任务的整个处理周期完成后，才会处理下一个任务。

内容的提问来源于stack exchange，提问作者liaolunhui