再现AKKA流异步输出

Question

Source(1 to 3)
  .map { i => println(s"A: $i"); i }
  .map { i => println(s"B: $i"); i }
  .map { i => println(s"C: $i"); i }
  .runWith(Sink.ignore)

A: 1
A: 2
B: 1
A: 3
B: 2
C: 1
B: 3
C: 2
C: 3

Answer 1

您之所以看到顺序操作，是因为您的所有操作都来自同一个源，因此在同一个异步边界内。要获得您正在寻找的“异步行为”，您需要添加流：

implicit val actorSystem = ActorSystem()
implicit val actorMaterializer = ActorMaterializer()

Source(1 to 3).via(Flow[Int].map{i => println(s"A: $i"); i })
              .via(Flow[Int].map{i => println(s"B: $i"); i })
              .via(Flow[Int].map{i => println(s"C: $i"); i })
              .runWith(Sink.ignore)

每个流将实现为一个独立的执行元。注意：若要获得真正的并发性，ActorSystem正在操作的线程池必须有1个以上的线程。

需要记住的一点是：ActorSystem的好处是它负责操作的低层控制，这样开发人员就可以专注于“业务逻辑”。这也可能是一个缺点。根据ActorSystem配置、JVM配置和硬件配置，操作顺序可能仍然是同步的。