ForkJoinPool 用于并行处理答案

【问题标题】：ForkJoinPool for parallel processingForkJoinPool 用于并行处理
【发布时间】：2015-07-28 04:13:12
【问题描述】：

我正在尝试运行一些代码 100 万次。我最初使用 Threads 编写它，但这似乎很笨拙。我开始做更多的阅读，我遇到了ForkJoin。这似乎正是我所需要的，但我不知道如何将下面的内容翻译成“scala 风格”。有人可以解释在我的代码中使用 ForkJoin 的最佳方式吗？

val l = (1 to 1000000) map {_.toLong}
println("running......be patient")
l.foreach{ x =>
    if(x % 10000 == 0) println("got to: "+x)
    val thread = new Thread {
        override def run { 
         //my code (API calls) here. writes to file if call success
        }
    }
}

【问题讨论】：

好的，有人填我。(1L to 1000000) 不是比事后申请map(_.toLong) 更有效率吗？
那很有可能，我今天开始学习Scala，所以我的代码肯定没有优化。

标签： multithreading scala fork-join

【解决方案1】：

最简单的方法是使用par（它会自动使用ForkJoinPool）：

 val l = (1 to 1000000) map {_.toLong} toList

 l.par.foreach { x =>
    if(x % 10000 == 0) println("got to: " + x) //will be executed in parallel way
    //your code (API calls) here. will also be executed in parallel way (but in same thread with `println("got to: " + x)`)
 }

另一种方式是使用Future:

import scala.concurrent._
import ExecutionContext.Implicits.global //import ForkJoinPool

val l = (1 to 1000000) map {_.toLong}

println("running......be patient")

l.foreach { x =>
    if(x % 10000 == 0) println("got to: "+x)
    Future {
       //your code (API calls) here. writes to file if call success
    }
}

如果您需要窃取工作 - 您应该使用 scala.concurrent.blocking 标记阻塞代码：

Future {
   scala.concurrent.blocking {
      //blocking API call here
   }
}

它会告诉ForkJoinPool 用新的线程来补偿阻塞的线程——这样你就可以避免线程饥饿（但有一些disadvantages）。

【讨论】：

感谢您的详细解答！如果每个 API 调用彼此完全独立，是否有理由在这种情况下使用 Future 而不是 ForkJoinPool？
ForkJoinPool 是 ThreadPool，Future 在其上被物理执行。 Future 是一种抽象，用于为 Java 的 ThreadPools 创建任务（您可以手动完成，但未来会更容易）。您可以在互联网上阅读更多关于 Futures 的信息，但关键优势是您可以轻松地以非阻塞方式提取计算结果（更多-更少）。 + 您可以自动支持池（如果您手动创建线程，则不需要），包括 ForkJoin 作为特殊情况（scala 默认选择它）。

【解决方案2】：

在 Scala 中，您可以使用 Future and Promise:

val l = (1 to 1000000) map {
  _.toLong
}
println("running......be patient")
l.foreach { x =>
  if (x % 10000 == 0) println("got to: " + x)
  Future{
    println(x)
  }
}

【讨论】：