一、任務分解問題和ForkJoinPool簡介

       在多線程并發編程中，有時候會遇到將大任務分解成小任務再并發執行的場景。Java 8新增的ForkJoinPool很好的支持了這個問題。

       ForkJoinPool是一種支持任務分解的線程池，當提交給他的任務“過大”，他就會按照預先定義的規則將大任務分解成小任務，多線程并發執行。

      一般要配合可分解任務接口ForkJoinTask來使用，ForkJoinTask有兩個實現它的抽象類：RecursiveAction和RecursiveTask，其區別是前者沒有返回值，后者有返回值。

          下面通過具體代碼，來示范兩個問題：（1）怎么定義可分解的任務類 （2）如何使用ForkJoinPool

package demo.thread.fork;

import java.util.Random;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.Future;

import java.util.concurrent.RecursiveTask;

public class ForkJoinPoolDemo {

public static void main(String[] args) throws InterruptedException, ExecutionException {

int arr[] = new int[100];

Random random = new Random();

int total = 0;

// 初始化100個數字元素

for (int i = 0; i < arr.length; i++) {

int temp = random.nextInt(100);

// 對數組元素賦值,并將數組元素的值添加到total總和中

total += (arr[i] = temp);

}

System.out.println("初始化時的總和=" + total);

// 創建包含Runtime.getRuntime().availableProcessors()返回值作為個數的并行線程的ForkJoinPool

ForkJoinPool forkJoinPool = new ForkJoinPool();

// 提交可分解的PrintTask任務

Future<Integer> future = forkJoinPool.submit(new SumTaskDemo(arr, 0,

arr.length));

System.out.println("計算出來的總和=" + future.get());

// 關閉線程池

forkJoinPool.shutdown();

}

}

// RecursiveTask為ForkJoinTask的抽象子類，有返回值的任務

class SumTaskDemo extends RecursiveTask<Integer> {

private static final long serialVersionUID = 4033241174438751063L;

// 每個"小任務"最多只打印20個數

private static final int MAX = 20;

private int arr[];

private int start;

private int end;

SumTaskDemo(int arr[], int start, int end) {

this.arr = arr;

this.start = start;

this.end = end;

}

@Override

protected Integer compute() {

int sum = 0;

// 當end-start的值小于MAX時候，開始打印

if ((end - start) <= MAX) {

for (int i = start; i < end; i++) {

sum += arr[i];

}

return sum;

} else {

System.err.println("=====任務分解======");

// 將大任務分解成兩個小任務

int middle = (start + end) >>>1;

SumTaskDemo left = new SumTaskDemo(arr, start, middle);

SumTaskDemo right = new SumTaskDemo(arr, middle, end);

// 并行執行兩個小任務

left.fork();

right.fork();

// 把兩個小任務累加的結果合并起來

return left.join() + right.join();

}

}

}
例子二：

package demo.thread.fork;

import java.util.Random;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.RecursiveAction;

import java.util.concurrent.TimeUnit;

/**

 * @author lin

 *

 */

public class ForkJoinPoolDemo2 {

public static void main(String[] args) throws Exception {

// 創建一個支持分解任務的線程池ForkJoinPool

ForkJoinPool pool = new ForkJoinPool(4);

myTask task = new myTask(60);

pool.submit(task);

pool.awaitTermination(10, TimeUnit.SECONDS);// 等待20s，觀察結果

pool.shutdown();

}

}

/**

 * 定義一個可分解的的任務類，繼承了RecursiveAction抽象類 必須實現它的compute方法

 */

class myTask extends RecursiveAction {

private static final long serialVersionUID = 1L;

// 定義一個分解任務的閾值——50,即一個任務最多承擔50個工作量

int THRESHOLD = 20;

// 任務量

int task_Num = 0;

myTask(int Num) {

this.task_Num = Num;

}

@Override

protected void compute() {

if (task_Num <= THRESHOLD) {

System.out.println(Thread.currentThread().getName() + "承擔了"

+ task_Num + "份工作");

/*try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}*/

} else {

// 隨機解成兩個任務

Random m = new Random();

int x = m.nextInt(50);

myTask left = new myTask(x);

myTask right = new myTask(task_Num - x);

left.fork();

right.fork();

}

}

}