1概要

fork/joinフレームワークはJava 7で提供されました。これは利用可能なすべてのプロセッサコアを使用しようとすることで並列処理を高速化するためのツールを提供します。

実際には、これは、フレームワークが最初に「フォーク」し、非同期に実行するのに十分単純になるまでタスクを小さい独立したサブタスクに再帰的に分割することを意味します。

その後、

「join」部分が始まり、

すべてのサブタスクの結果が再帰的に単一の結果に結合されます。または、voidを返すタスクの場合、プログラムは単にすべてのサブタスクが実行されるまで待機します。

効果的な並列実行を提供するために、fork/joinフレームワークは

ForkJoinWorkerThread

型のワーカースレッドを管理する

ForkJoinPool

と呼ばれるスレッドのプールを使用します。


2

フォークジョインプール



ForkJoinPool

はフレームワークの中心です。これは、ワーカースレッドを管理し、スレッドプールの状態とパフォーマンスに関する情報を取得するためのツールを提供する


ExecutorService


の実装です。

ワーカースレッドは一度に1つのタスクしか実行できませんが、

ForkJoinPool

はサブタスクごとに別々のスレッドを作成するわけではありません。代わりに、プール内の各スレッドは、タスクを格納する独自の両端キュー（またはhttps://en.wikipedia.org/wiki/Double-ended__queue[deque]）を持っています。

このアーキテクチャは、** work-stealingアルゴリズムを使用してスレッドのワークロードのバランスをとるために不可欠です。


2.1. 作業盗用アルゴリズム

  • 単純に言うと、空きスレッドはビジー状態のスレッドのデキューから作業を「盗む」ことを試みます。

デフォルトでは、ワーカースレッドはそれ自身の両端キューの先頭からタスクを取得します。

空の場合、スレッドは別のビジー状態のスレッドの両端キューの末尾またはグローバルエントリキューからタスクを引き継ぎます。これが、ここで最大の作業が行われる可能性が高いからです。

この方法では、スレッドがタスクを競合する可能性が最小限に抑えられます。それはまた、スレッドが最初に利用可能な最大のチャンクを処理するので、スレッドが仕事を探しに行かなければならなくなる回数を減らします。


2.2.

ForkJoinPool



インスタンス化

Java 8では、

ForkJoinPool

のインスタンスにアクセスする最も便利な方法は、その静的メソッド

（）を使用することです。すべての

ForkJoinTask__。


Oracleのドキュメント

によれば、事前定義された共通プールを使用するとリソースの消費が削減されます。タスクごとに別々のスレッドプール。 

ForkJoinPool commonPool = ForkJoinPool.commonPool();

Java 7でも

ForkJoinPool

を作成し、それをユーティリティクラスの

public static

フィールドに割り当てることで同じ動作を実現できます。 

public static ForkJoinPool forkJoinPool = new ForkJoinPool(2);

これで簡単にアクセスできます。 

ForkJoinPool forkJoinPool = PoolUtil.forkJoinPool;

フォークジョインプールのコンストラクタを使用すると、特定レベルの並列処理、スレッドファクトリ、および例外ハンドラを使用してカスタムスレッドプールを作成できます。上記の例では、プールの並列処理レベルは2です。これは、プールが2つのプロセッサコアを使用することを意味します。


3

ForkJoinTask <V>



ForkJoinTask

は、

ForkJoinPoolの中で実行されるタスクの基本型です。実際には、その2つのサブクラスのうちの1つを拡張する必要があります。両方ともタスクのロジックが定義される抽象メソッド

compute（）__を持っています。

** 3.1. __RecursiveAction – 例+

__ **

以下の例では、処理される作業単位は

workload

という

String

で表されています。デモンストレーションの目的のために、タスクは無意味なものです:それは単にその入力を大文字にしてそれを記録します。

フレームワークの分岐動作を説明するために、

workload.length（）

が

createSubtask（）

メソッドを使用して指定されたしきい値


より大きい場合、例でタスクを分割します。

Stringは再帰的に部分文字列に分割され、これらの部分文字列に基づく

CustomRecursiveTask

インスタンスが作成されます。

結果として、メソッドは

List <CustomRecursiveAction> .

を返します。

リストは

invokeAll（）

メソッドを使用して

ForkJoinPool

に送信されます。 

public class CustomRecursiveAction extends RecursiveAction {

    private String workload = "";
    private static final int THRESHOLD = 4;

    private static Logger logger =
      Logger.getAnonymousLogger();

    public CustomRecursiveAction(String workload) {
        this.workload = workload;
    }

    @Override
    protected void compute() {
        if (workload.length() > THRESHOLD) {
            ForkJoinTask.invokeAll(createSubtasks());
        } else {
           processing(workload);
        }
    }

    private List<CustomRecursiveAction> createSubtasks() {
        List<CustomRecursiveAction> subtasks = new ArrayList<>();

        String partOne = workload.substring(0, workload.length()/2);
        String partTwo = workload.substring(workload.length()/2, workload.length());

        subtasks.add(new CustomRecursiveAction(partOne));
        subtasks.add(new CustomRecursiveAction(partTwo));

        return subtasks;
    }

    private void processing(String work) {
        String result = work.toUpperCase();
        logger.info("This result - (" + result + ") - was processed by "
          + Thread.currentThread().getName());
    }
}

これを行うには、総作業量を表すオブジェクトを作成し、適切なしきい値を選択し、作業を分割する方法を定義し、作業を実行する方法を定義します。


3.2.

RecursiveTask <V>


値を返すタスクの場合、ここに示すロジックは似ていますが、各サブタスクの結果は単一の結果にまとめられます。 

public class CustomRecursiveTask extends RecursiveTask<Integer> {
    private int[]arr;

    private static final int THRESHOLD = 20;

    public CustomRecursiveTask(int[]arr) {
        this.arr = arr;
    }

    @Override
    protected Integer compute() {
        if (arr.length > THRESHOLD) {
            return ForkJoinTask.invokeAll(createSubtasks())
              .stream()
              .mapToInt(ForkJoinTask::join)
              .sum();
        } else {
            return processing(arr);
        }
    }

    private Collection<CustomRecursiveTask> createSubtasks() {
        List<CustomRecursiveTask> dividedTasks = new ArrayList<>();
        dividedTasks.add(new CustomRecursiveTask(
          Arrays.copyOfRange(arr, 0, arr.length/2)));
        dividedTasks.add(new CustomRecursiveTask(
          Arrays.copyOfRange(arr, arr.length/2, arr.length)));
        return dividedTasks;
    }

    private Integer processing(int[]arr) {
        return Arrays.stream(arr)
          .filter(a -> a > 10 && a < 27)
          .map(a -> a **  10)
          .sum();
    }
}

この例では、作業は

CustomRecursiveTask

クラスの

arr

フィールドに格納されている配列によって表されます。

createSubtask（）

メソッドは、各部分がしきい値より小さくなるまで、タスクを再帰的に小さな作業に分割します。次に、

invokeAll（）

メソッドは、サブタスクを共通プルに送信し、


https://docのリストを返します。

oracle.com/javase/8/docs/api/java/util/concurrent/Future.html[Future]

実行をトリガーするために、

join（）

メソッドが各サブタスクに対して呼び出されました。

この例では、これはJava 8の


Stream API

;

the

sum（）

を使用して実現されます。 methodは、サブ結果を最終結果に結合する表現として使用されます。


4

ForkJoinPool


へのタスクの送信

タスクをスレッドプールに送信するには、いくつかの方法を使用できます。


  • submit（）


    または



    execute（）** メソッド（それらのユースケースは同じ） 

forkJoinPool.execute(customRecursiveTask);
int result = customRecursiveTask.join();



invoke（）


メソッドはタスクをフォークして結果を待ちます。手動での参加は必要ありません。 

int result = forkJoinPool.invoke(customRecursiveTask);


  • invokeAll（）

    ** メソッドは

    ForkJoinTasks

    のシーケンスを__ForkJoinPoolに渡す最も便利な方法です。それらが生成された順序。

あるいは、別々の


fork（）

メソッドと

join（）


メソッドを使用することもできます。

fork（）

メソッドはタスクをプールに送信しますが、実行をトリガーすることはありません。このために

join（）

メソッドを使用します。

RecursiveAction

の場合、

join（）

は

null

以外の何も返しません。

RecursiveTask <V>の場合、

はタスクの実行結果を返します。 

customRecursiveTaskFirst.fork();
result = customRecursiveTaskLast.join();


RecursiveTask <V>

の例では、

invokeAll（）

メソッドを使用して、一連のサブタスクをプールに送信しました。

fork（）

と

join（）

でも同じことができますが、結果の順序付けに影響があります。

混乱を避けるために、

invokeAll（）

メソッドを使用して__ForkJoinPoolに複数のタスクを送信することをお勧めします。


5結論

fork/joinフレームワークを使用すると、大きなタスクの処理速度を上げることができますが、この結果を達成するには、いくつかのガイドラインに従う必要があります。


  • できるだけ少ないスレッドプールを使用する

    – ほとんどの場合、最高のスレッドプール

アプリケーションまたはシステムごとに1つのスレッドプールを使用するという決定


特別な調整が必要ない場合は、** デフォルトの共通スレッドプールを使用します。


  • 分岐フォークタスクをに分割するための妥当なしきい値を使用する

サブタスク


あなたの



ForkJoingTasks

** でブロッキングを避ける

この記事で使用されている例はhttps://github.com/eugenp/tutorials/tree/master/core-java-concurrency[linked GitHub repository]にあります。