平行集合

設定平行集合

語言

任務支援

平行集合在作業排程的方式上是模組化的。每個平行集合都會參數化一個任務支援物件,負責將任務排程和負載平衡到處理器。

任務支援物件在內部保留對執行緒池實作的參考,並決定如何以及何時將任務分割成較小的任務。若要深入了解其執行方式,請參閱技術報告 [1]

目前有幾個任務支援實作可供平行集合使用。 ForkJoinTaskSupport 在內部使用 fork-join 池,並在 JVM 1.6 或更新版本中預設使用。效率較低的 ThreadPoolTaskSupport 是 JVM 1.5 和不支援 fork join 池的 JVM 的後備選項。 ExecutionContextTaskSupport 使用 scala.concurrent 中找到的預設執行緒內容實作,並重複使用 scala.concurrent 中使用的執行緒池(這可能是 fork join 池或執行緒池執行器,視 JVM 版本而定)。執行緒內容任務支援預設設定至每個平行集合,因此平行集合重複使用與未來 API 相同的 fork-join 池。

以下是變更平行集合任務支援的方法

scala> import scala.collection.parallel._
import scala.collection.parallel._

scala> val pc = mutable.ParArray(1, 2, 3)
pc: scala.collection.parallel.mutable.ParArray[Int] = ParArray(1, 2, 3)

scala> val forkJoinPool = new java.util.concurrent.ForkJoinPool(2)
forkJoinPool: java.util.concurrent.ForkJoinPool = java.util.concurrent.ForkJoinPool@6436e181[Running, parallelism = 2, size = 0, active = 0, running = 0, steals = 0, tasks = 0, submissions = 0]

scala> pc.tasksupport = new ForkJoinTaskSupport(forkJoinPool)
pc.tasksupport: scala.collection.parallel.TaskSupport = scala.collection.parallel.ForkJoinTaskSupport@4a5d484a

scala> pc map { _ + 1 }
res0: scala.collection.parallel.mutable.ParArray[Int] = ParArray(2, 3, 4)

上述設定將平行集合設定為使用並行等級 2 的 fork-join 池。若要設定平行集合使用執行緒池執行器

scala> pc.tasksupport = new ThreadPoolTaskSupport()
pc.tasksupport: scala.collection.parallel.TaskSupport = scala.collection.parallel.ThreadPoolTaskSupport@1d914a39

scala> pc map { _ + 1 }
res1: scala.collection.parallel.mutable.ParArray[Int] = ParArray(2, 3, 4)

scala> forkJoinPool.shutdown()

請注意,如果您正在建立自己的 ForkJoinPool 執行個體,您應該在不再需要執行緒池時呼叫 ForkJoinPool.shutdown()。如果您未呼叫 ForkJoinPool.shutdown() 並繼續建立 ForkJoinPool 的新執行個體,JVM 最終可能會用盡可用的執行緒並擲出 java.lang.OutOfMemoryError

當平行集合序列化時,任務支援欄位會從序列化中省略。當反序列化平行集合時,任務支援欄位會設定為預設值,即執行緒內容任務支援。

若要實作自訂工作支援,請延伸 TaskSupport 特性並實作下列方法

def execute[R, Tp](task: Task[R, Tp]): () => R

def executeAndWaitResult[R, Tp](task: Task[R, Tp]): R

def parallelismLevel: Int

execute 方法非同步地排程工作並傳回一個未來,以等待運算結果。 executeAndWait 方法執行相同的動作,但只在工作完成時傳回。 parallelismLevel 僅傳回工作支援用於排程工作的目標核心數目。

參考

  1. 通用平行收集架構,2011 年 6 月

此頁面的貢獻者