2010-02-03 95 views
4

我想在斯卡拉使用并发编程。基于this example这里在 StackOverflow,我做了一个程序根据项目欧拉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)
  • 期货:...我在3分钟后取消执行

很明显,我无法正确使用来自Java和Scala的并发API。所以我问: 我的错误在哪里?什么是使用并发的更合适的形式? 关于Scala演员?有可能使用它们吗?

回答

1

你期待什么结果?你是否期望这些方法中的一种比其他方法更好?你是否期望程序针对不同的执行方法进行不同的缩放?

你的机器有多少核心?如果你只有一个核心,那么你应该预计时间会随着工作线性增加。在运行过程中,你的CPU使用情况如何?数字是否可重复?

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

1

我假设你使用的是Scala 2.7。基本上,filtermapRange1 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范围,它会没事的。