理解力与Future.sequence

Question

我的机器上有四个核心。

我认为，为了更好地理解日程安排(不需要依赖于以前的result..same，如flatMap)：

val stuffResult: Future[String] = for {
   stuff1 <- service1.doServiceStuff("service 1")  // worker-5 (4+1)
   stuff2 <- service1.doServiceStuff("service 2")
   stuff3 <- service1.doServiceStuff("service 3")
   stuff4 <- service1.doServiceStuff("service 4")
 } yield (stuff1 + ", " + stuff2 + ", "+ stuff3 + ", " + stuff4)

哪里

 class Service {

 implicit val blockingExContext = scala.concurrent.ExecutionContext.fromExecutor(null: Executor)

 def doServiceStuff(name:String): Future[String] = {
    Future {
      blocking {    
        println ( s"start ${name} on " + Thread.currentThread.getName)
        Thread.sleep(5000)
        "stuff_done"
      }
    }
  }
}

但我看到的是(每一步都要花费大约5秒)：

• 开始服务1在ForkJoinPool-3-工人-5
• ForkJoinPool 2启动服务-3-工人-5
• 在ForkJoinPool上启动服务3-工人-5
• ERROR:java.util.concurrent.TimeoutException:期货在10秒后超时

所有线程都运行在一个线程上，而不是使用现有的空闲线程，而是尽可能快地在5秒内完成所有线程。

但如果我说：

val stuff1 = service1.doServiceStuff("service 1")
val stuff2 = service1.doServiceStuff("service 2")
val stuff3 = service1.doServiceStuff("service 3")
val stuff4 = service1.doServiceStuff("service 4")

Future.sequence(List(stuff1, stuff2, stuff3, stuff4)).map { list =>
  list.foldLeft("") { (acc, x) => acc + " " + x }
}
..

一切在5秒内结束。

理解力在哪种情况下是依次工作的？是吗？

Answer 1

它不能按顺序工作，它只是在创建Futures之前不能启动它们(在您的上一个Future的flatMap中会发生这种情况)，所以如果您想要并行地处理它们(使用通常的隐式ExecutionContext)，则需要提前创建它们。

不过，本教程可能解释得更好(它使withFilter使其复杂化)：

购买期货只有在usdQuote和chfQuote完成之后才能完成--它取决于这两个期货的价值，因此它自己的计算不能提前开始。 以上的理解被翻译成： val purchase = usdQuote flatMap { usd => chfQuote .withFilter(chf => isProfitable(usd, chf)) .map(chf => connection.buy(amount, chf)) } 这比理解要难一些，但是我们分析它是为了更好地理解flatMap操作。flatMap操作将自己的价值映射到其他一些未来。一旦完成了这个不同的未来，结果的未来就会以其价值完成。在我们的示例中，flatMap使用usdQuote期货的值将chfQuote的值映射到第三个未来，后者发送购买一定金额的瑞士法郎的请求。由此产生的未来购买只有在地图返回的第三个未来完成后才会完成。

您真正需要的只是类似于map2而不是flatMap的东西，因为您不使用来自前一个Future的返回值来创建新的Future。