Java Fork/Joinフレームワークについて

今日もN予備校で学習

タイトルはJavaだけどScalaの勉強で必要になったので(^_^;)

記事内のコードはScalaです。

Fork/Join フレームワーク

Executorフレームワークでは、スレッドは割り当てられたキューのタスク消化を行った(待機スレッドは他スレッドのタスクキューを消化できない)が、Fork/Joinフレームワークではプール内の待機スレッドは、プールに送信されたタスク、他のアクティブなタスクによって作成されたタスク、あるいはその両方を見つけて実行しようする。

ForkJoinPool (Java Platform SE 8)

ForkJoinTask

ForkJoinPool(Fork/Join フレームワークでのスレッドプール)ではForkJoinTaskインスタンスをタスクとして実行する。

ForkJoinTaskは以下の2種類の抽象クラスの実装となる。

• RecursiveTask: 結果を返す
• RecursiveAction: 結果を返さない

ForkJoinTaskの実装例。computeメソッドを実装する。引数listに処理対象の行列が入ってくるとする。computeは行列を適切に分割しながらタスクを処理する記述を行う。

class DoActionTask(list: List[BigInt]) extends RecursiveTask[BigInt] {

  override def compute(): BigInt = {
    val n = list.length / 2
    if (n == 0) {
      list match {
        case List() => 0
        case List(n) => doAction(n)
      }
    } else {
      val (left, right) = list.splitAt(n)
      val leftTask = new DoActionTask(left)
      val rightTask = new DoActionTask(right)
      leftTask.fork()
      rightTask.fork()
      leftTask.join() + rightTask.join()
    }
  }
}

ここでは省略しているがdoActionに行列に対しての処理を記述する。

基本、再帰で処理分割を行っていくのだが、forkで別スレッドで実行可能なcomputeメソッドとして分割できる(表現が合ってるかは謎...)。 forkでの実行結果はjoinメソッドで受け取れる。結果はcomputeメソッドが再帰された結果となる。

タスクの実行

タスクの実行は、以下の3種類がある。

• execute: 非同期実行を行う
• invoke: 同期実行を行う
• submit: 実行を行い Future のインスタンスを取得する

使用例

import java.util.concurrent.{ForkJoinPool, RecursiveTask}
val pool = new ForkJoinPool()
val result = pool.invoke(new AggregateTask(list))

リンク

リンク