Scala并行集合框架初探

Scala并行集合框架初探

1 并行集合框架简介

 Scala 并行集合框架( Parallel Collections Framework)是在2.9版添加的重要功能，用于多核环境的并行计算。

主要用到的算法有：

 

• divide and conquer : 分治算法

   Scala通过splitters，combiners等抽象层来实现，主要原理是将计算工作分解很多任务，分发给一些处理器去完成[,并将它们处理结果合并返回]。

 

• Work stealing算法

   主要用于任务调度负载均衡（load-balancing），通俗点完成自己的所有任务之后，发现其他人还有活没干完，主动（或被安排）帮他人一起干，这样达到尽早干完的目的。

 

并行集合位于scala.collection.parallel，跟普通集合(regular collections)一样,分immutable和mutable。主要实现类是：

 

scala.collection.parallel.mutable.ParArray
scala.collection.parallel.mutable.ParHashMap
scala.collection.parallel.mutable.ParHashSet
	 
scala.collection.parallel.immutable.ParRange
scala.collection.parallel.immutable.ParHashMap
scala.collection.parallel.immutable.ParHashSet
scala.collection.parallel.immutable.ParVector

 

 

基本上命名是对应普通集合名加"Par"前缀。

 

2 常见用法

2.1 构造并行集合

像普通集合一样，并行集合都有相对应的object工厂对象来构造集合实例。以ParVector为例：

 

$ scala> import scala.collection.parallel.immutable.ParVector
$ scala> val a = ParVector(1,2)
            a: scala.collection.parallel.immutable.ParVector[Int] = ParVector(1, 2)
$ scala> a.size
            res0: Int = 2
$ scala> a == ParVector.apply(1,2)
            res1: Boolean = true

 

 

2.2 普通集合与并行集合的相互转换

Scala为普通集合类增加par实例方法，用于转换为对应的并行集合；同时，每个并行集合类也有seq实例方法，用于转换到对应的普通集合。如：

 

 

 

scala> List(1,2).par
        res0: scala.collection.parallel.immutable.ParSeq[Int] = ParVector(1, 2)

scala> List(1,2).par.seq
        res1: scala.collection.immutable.Seq[Int] = Vector(1, 2)

 

2.3 并行集合和普通集合的公用接口

它们都实现了GenIterable, GenSeq接口。如：

 

scala> import scala.collection.GenSeq
scala> def p[T](xs: GenSeq[T]) = xs foreach print
scala> p((1 to 5).toList)
        12345
 scala> p((1 to 5).toList.par)
        31245

 

 

2.4 并行集合的常用操作

Scala平滑聪明的实现并行集合，并与普通集合公用一致的接口。所以大部分操作就行使用普通集合一样，如foreach, map, flatMap, filter 等。见以下示例：

 

scala> (1 to 5).par.foreach{it => println(Thread.currentThread);print("^"+ it)}
        Thread[ForkJoinPool-1-worker-1,5,main]
        Thread[ForkJoinPool-1-worker-3,5,main]
        ^4Thread[ForkJoinPool-1-worker-3,5,main]
        ^5Thread[ForkJoinPool-1-worker-0,5,main]
        ^1Thread[ForkJoinPool-1-worker-3,5,main]
        ^2^3
scala> (1 to 5).par.map( _ + 100)
        res5: scala.collection.parallel.immutable.ParSeq[Int] = ParVector(101, 102, 103, 104, 105)

scala> List(1,2,3).par.filter(_ % 2 == 0)
        res6: scala.collection.parallel.immutable.ParSeq[Int] = ParVector(2)

 

 

 这里要注意的是foreach,是在多个线程下执行，不保证迭代元素顺序了。

 

3 实践

3.1 普通集合和并行集合的性能简单比较

环境: 本人notebook：Fedora 64位；内存4G；4核CPU

JDK： 1.6.0_26，64-Bit Server VM

Scala: 2.9.0.1，启动参数：scala -J-Xms1200M -J-Xmx2000M

代码如下：

 

object ParBenchmark {
      case class Result(item: Int, c: Long, p: Long) {
        override def toString = "%-10s\t%-10d\t%-10d".format(item, c, p)
      }
      def time(proc: => Any) = {
        def curr = System.currentTimeMillis
        val s = curr; proc; curr - s
      }
      def even(i: Int) = i % 2 == 0

      def b(count: Int) = Some((1 to count).toList).
        map(it => (it, it.par)).headOption.
        map { it =>
          Result(it._1.size, time(it._1 filter even), time(it._2 filter even))
        }
      def main(args: Array[String]): Unit = {
        println("item\tcommon\tpar")
        Array(1, 2, 5, 10, 12, 15, 18, 20).map(_ * 1000000).
          foreach { it =>
            Runtime.getRuntime.gc()
            Thread.sleep(2000)

            println(b(it).get)
          }
      }
    }

 

几次执行结果如下：

 

item regular par
1000000 36 57
2000000 29 24
5000000 70 926
10000000 133 87
12000000 3381 99
15000000 200 124
18000000 1363 162
20000000 1273 219

item regular par
1000000 32 56
2000000 29 21
5000000 71 928
10000000 134 91
12000000 3390 110
15000000 205 127
18000000 1269 168
20000000 1169 196

 

 

 

初步结论

 

• 计算数据量少，并行集合性能不占优势，甚至还处于劣势

             估计是线程切换、分发合并等额外操作消耗的时间。

       

 

• 计算数据量大时，如达到千万级别时，并行集合性能优势凸显出来了

             当然这些跟本机硬件环境相关，CPU数内核数越多，并行计算当然更有效率。

 

4 更深入

深入理解并行集合框架实现的细节，学习《A Generic Parallel Collection Framework》论文；

与JDK 7 fork/join 框架的关系及对比；

5 附录

更多使用见测试用例代码：

  https://github.com/itang/_learn/blob/master/scala/what_is_new_scala_2.9.0/src/test/scala/me/itang/learn_scala/what_is_new_scala_290/ParallelSpec.scala

 

参考资料：

  http://kotek.net/blog/quick_look_at_upcoming_parallel_collections_in_scala_2.9

  http://infoscience.epfl.ch/record/150220/files/pc.pdf