Scala 中的并发处理答案

【问题标题】：Concurrent Processing in ScalaScala 中的并发处理
【发布时间】：2010-02-03 01:52:38
【问题描述】：

我正在尝试在 Scala 中使用并发编程。基于this example这里 StackOverflow，我基于Project Euler的Problem 1做了一个程序。我尝试了三种方法：第一种是简单的执行，没有并行性。这其次通过 Executors 和 Callables 使用 java.util.concurrency API。第三，基于上面提到的页面，使用scala.Futures。我的目标是比较执行时间。

这是代码：

package sandbox

import java.util.concurrent._
import scala.actors._

object TestPool {

  def eval(n: Int): Boolean = (n % 3 == 0) || (n % 5 == 0)

  def runSingle(max: Int): Int = (1 until max).filter(eval(_)).foldLeft(0)(_ + _)

  def runPool(max: Int): Int = {

    def getCallable(i: Int): Callable[Boolean] = new Callable[Boolean] { def call = eval(i) }

    val pool = Executors.newFixedThreadPool(5)
    val result = (1 until max).filter(i => pool.submit(getCallable(i)).get).foldLeft(0)(_ + _)
    pool.shutdown
    pool.awaitTermination(Math.MAX_LONG, TimeUnit.SECONDS)

    result
  }

  def runFutures(max: Int): Int = (1 until max).filter(i => Futures.future(eval(i)).apply).foldLeft(0)(_ + _)

  /**
   * f is the function to be runned. it returns a Tuple2 containing the sum and the 
   * execution time.
   */
  def test(max: Int, f: Int => Int): (Int, Long) = {
    val t0 = System.currentTimeMillis
    val result = f(max)
    val deltaT = System.currentTimeMillis - t0

    (result, deltaT)
  }


  def main(args : Array[String]) : Unit = {
    val max = 10000

    println("Single : " + test(max, runSingle))
    println("Pool   : " + test(max, runPool))
    println("Futures: " + test(max, runFutures))
  }
}

这些是结果：

最大值 = 10：

单人：(23,31)
池：(23,16)
期货：(23,31)

最大值 = 100：

单人：(2318,33)
池：(2318,31)
期货：(2318,55)

最大值 = 1000：

单身：(233168,42)
池：(233168,111)
期货：(233168,364)

最大值 = 10000：

单身：(23331668,144)
池：(23331668,544)
Futures：……我在 3 分钟后取消了执行

显然我无法正确使用 Java 和 Scala 的并发 API。所以我问：我的错误在哪里？使用并发的更合适的形式是什么？关于 Scala Actors？可以用吗？

【问题讨论】：

标签： performance scala concurrency

【解决方案1】：

您期待什么结果？您是否期望其中一种方法比其他方法表现更好？您是否希望程序针对不同的执行方法进行不同的扩展？

您的机器有多少个内核？如果您只有一个核心，那么您应该预期时间会随着要做的工作线性增加。在运行过程中，您的 cpu 使用情况如何？数字可以重复吗？

您还没有考虑到 JVM Hotspot 预热时间的影响，这可能会导致像这样的微基准测试出现严重问题。

【讨论】：

【解决方案2】：

我假设您使用的是 Scala 2.7。基本上，Range 上的 filter 和 map（1 until max 的结果）是非严格的，这意味着它将按需计算，并且每次您尝试访问它的结果时都会计算它。

试试这个，例如：

val y = (1 to 10).filter{x => println("Filtering "+x); x % 2 == 0}.map{x => println("Mapping "+x); x * 2}
println(y(0))
println(y(1))
println(y(2))
println(y(0))

无论如何，结果是您的并行内容是串行运行的。在范围内添加一个.force，就可以了。

【讨论】：