Fork/Join使用学习

JDK7提供了一个将任务“分而治之”的框架 — Fork/Join。它把一个大的任务分割成足够小的子任务,如果子任务比较大的话还要对子任务进行继续分割。分割的子任务分别放到双端队列里,然后启动线程分别从双端队列里获取任务执行。子任务执行完的结果都放在另外一个队列里,启动一个线程从队列里取数据,然后合并这些数据。

Fork/Join的思想如下所示: QQ截图20190702172347.png

RecursiveTask

QQ截图20190703111904.png

RecursiveTask适用于将任务分而治之,并且有返回值的情况,举个计算1到100和的例子:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
public class RecursiveTest {
// 定义最小区间为10
private final static int MAX_THRESHOLD = 10;

public static void main(String[] args) {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
ForkJoinTask<Integer> future = forkJoinPool.submit(new CalculateRecursiveTask(1, 100));
try {
Integer result = future.get();
System.out.println(result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}

private static class CalculateRecursiveTask extends RecursiveTask<Integer> {
// 起始
private int start;
// 结束
private int end;

public CalculateRecursiveTask(int start, int end) {
this.start = start;
this.end = end;
}

@Override
protected Integer compute() {
// 如果起始和结束范围小于我们定义的区间范围,则直接计算
if ((end - start) <= MAX_THRESHOLD) {
return IntStream.rangeClosed(start, end).sum();
} else {
// 否则,将范围一分为二,分成两个子任务
int middle = (start + end) / 2;
CalculateRecursiveTask leftTask = new CalculateRecursiveTask(start, middle);
CalculateRecursiveTask rightTask = new CalculateRecursiveTask(middle + 1, end);
// 执行子任务
leftTask.fork();
rightTask.fork();

// 汇总子任务
return leftTask.join() + rightTask.join();
}
}
}
}

ForkJoinPool使用submit或invoke提交的区别:invoke是同步执行,调用之后需要等待任务完成,才能执行后面的代码;submit是异步执行,只有在Future调用get的时候会阻塞。

启动程序输出如下:

QQ截图20190703112347.png

其实这里执行子任务调用fork方法并不是最佳的选择,最佳的选择是invokeAll方法:

1
2
3
4
5
6
// 执行子任务
// leftTask.fork();
// rightTask.fork();
invokeAll(leftTask,rightTask);
// 汇总子任务
return leftTask.join() + rightTask.join();

RecursiveAction

QQ截图20190703114411.png

使用方式和RecursiveTask类似,只不过没有返回值:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public class RecursiveActionTest {
// 定义最小区间为10
private final static int MAX_THRESHOLD = 10;
private final static AtomicInteger SUM = new AtomicInteger(0);

public static void main(String[] args) throws InterruptedException {
final ForkJoinPool forkJoinPool = new ForkJoinPool();
forkJoinPool.submit(new CalculateRecursiveAction(0, 100));
forkJoinPool.awaitTermination(2, TimeUnit.SECONDS);
System.out.println(SUM);
}

private static class CalculateRecursiveAction extends RecursiveAction {
// 起始
private final int start;
// 结束
private final int end;

private CalculateRecursiveAction(int start, int end) {
this.start = start;
this.end = end;
}

@Override
protected void compute() {
// 如果起始和结束范围小于我们定义的区间范围,则直接计算
if ((end - start) <= MAX_THRESHOLD) {
SUM.addAndGet(IntStream.rangeClosed(start, end).sum());
} else {
// 否则,将范围一分为二,分成两个子任务
int middle = (end + start) / 2;
CalculateRecursiveAction leftAction = new CalculateRecursiveAction(start, middle);
CalculateRecursiveAction rightAction = new CalculateRecursiveAction(middle + 1, end);
// 执行子任务
invokeAll(leftAction, rightAction);
// 没有汇总子任务结果过程,因为没有返回值。
}
}
}
}

输出结果也是5050。

什么时候用

上面只是为了演示Fork/Join的用法,实际是采用这种方式计算反而更加费时,因为切割任务,分配线程需要额外的开销。其实什么时候用不必太纠结,一个足够大的任务,如果采用Fork/Join来处理比传统处理方式快的话,那就毫不犹豫的选择它吧!

参考文章:https://www.imooc.com/article/24822

请作者喝瓶肥宅水~

TOP