一、任務(wù)分解問題和ForkJoinPool簡介

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

       ForkJoinPool是一種支持任務(wù)分解的線程池，當(dāng)提交給他的任務(wù)“過大”，他就會按照預(yù)先定義的規(guī)則將大任務(wù)分解成小任務(wù)，多線程并發(fā)執(zhí)行。

      一般要配合可分解任務(wù)接口ForkJoinTask來使用，F(xiàn)orkJoinTask有兩個實現(xiàn)它的抽象類：RecursiveAction和RecursiveTask，其區(qū)別是前者沒有返回值，后者有返回值。

          下面通過具體代碼，來示范兩個問題：（1）怎么定義可分解的任務(wù)類 （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個數(shù)字元素

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

int temp = random.nextInt(100);

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

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

}

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

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

ForkJoinPool forkJoinPool = new ForkJoinPool();

// 提交可分解的PrintTask任務(wù)

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

arr.length));

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

// 關(guān)閉線程池

forkJoinPool.shutdown();

}

}

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

class SumTaskDemo extends RecursiveTask<Integer> {

private static final long serialVersionUID = 4033241174438751063L;

// 每個"小任務(wù)"最多只打印20個數(shù)

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;

// 當(dāng)end-start的值小于MAX時候，開始打印

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

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

sum += arr[i];

}

return sum;

} else {

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

// 將大任務(wù)分解成兩個小任務(wù)

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

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

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

// 并行執(zhí)行兩個小任務(wù)

left.fork();

right.fork();

// 把兩個小任務(wù)累加的結(jié)果合并起來

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 {

// 創(chuàng)建一個支持分解任務(wù)的線程池ForkJoinPool

ForkJoinPool pool = new ForkJoinPool(4);

myTask task = new myTask(60);

pool.submit(task);

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

pool.shutdown();

}

}

/**

 * 定義一個可分解的的任務(wù)類，繼承了RecursiveAction抽象類 必須實現(xiàn)它的compute方法

 */

class myTask extends RecursiveAction {

private static final long serialVersionUID = 1L;

// 定義一個分解任務(wù)的閾值——50,即一個任務(wù)最多承擔(dān)50個工作量

int THRESHOLD = 20;

// 任務(wù)量

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() + "承擔(dān)了"

+ task_Num + "份工作");

/*try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}*/

} else {

// 隨機解成兩個任務(wù)

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();

}

}

}