JUC并发编程——ForkJoin与异步回调

ForkJoin （分支合并）

什么是ForkJoin

ForkJoin在JDK1.7出现 ，并行执行任务，在大数据量下，能够提高效率

讯飞星火提供的说法：

Forkjoin是一种并行计算的算法，用于将一个大任务分解为多个小任务，然后将这些小任务分配给不同的线程或进程来并行执行，最后再将结果合并。

在计算机科学中，Forkjoin通常用于实现基于分治策略的程序和数据结构，例如排序算法、图遍历算法、哈希表等。它可以有效地利用多核处理器的并行计算能力，提高程序的性能。

ForkJoin特点

• 工作窃取

ForkJoin将一个大任务分解为若干个小任务进行并行运算，假若现在有两个线程，A线程与B线程，当B线程执行完它的小任务后发现，A线程还未执行完A线程的小任务，则B线程可以偷取A线程的小任务执行，这样可以加快任务执行效率。

同时需要注意：分配给A线程与B线程的任务存储结构是双端队列，因此B线程可以从另一头窃取A线程的任务而不会导致A线程与B线程执行同一个任务而导致异常或错误。

ForkJoin的操作

示例为，计算1叠加到10亿，在本示例中，用了三种不同的方法进行计算，打印结果与计算时长。将ForkJoin封装到ForkJoinDemo类中

ForkJoinDemo类：

package ForkJoin;

import java.util.concurrent.RecursiveTask;

/**
 *
 * 求和计算的任务！
 * 如何使用forkJoin
 * 1、通过 forkJoinPool 来执行
 * 2、计算任务 forkJoinPool.execute(ForKJoinTask task)
 * 3、计算类要继承ForkJoinTask
 */
public class ForkJoinDemo extends RecursiveTask<Long> {
    private Long start;
    private Long end;

    // 临界值
    private Long temp = 1000L;

    public ForkJoinDemo(Long start,Long end){
        this.start = start;
        this.end = end;
    }

    // 计算方法  RecursiveTask接口中唯一抽象方法
    @Override
    protected Long compute() {
        if ((end-start)>temp){
            Long sum = 0L;
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        }else{
            // forkJoin  思想上非常像递归
            // 中间值
            long middle = (start+end)/2;
            ForkJoinDemo task1 = new ForkJoinDemo(start, middle);
            task1.fork();// 拆分任务，把任务压入线程队列
            ForkJoinDemo task2 = new ForkJoinDemo(middle+1, start);
            task2.fork();// 拆分任务，把任务压入线程队列
            return task1.join()+task2.join();
        }
    }
}
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

测试类：该类中分别用了三种方法去计算求和，普通for循环，ForkJoin以及Stream

package ForkJoin;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;

public class Test {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        test1();
        test2();
        test3();
    }

    // 基础计算方法
    public static void test1(){
        Long sum = 0l;
        long start = System.currentTimeMillis();
        for (Long i = 1L; i <= 10_0000_0000; i++) {
            sum += i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"时间"+(end-start));
    }

    // forkJoin
    public static void test2() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> task = new ForkJoinDemo(1L, 10_0000_0000L);
        // forkJoinPool.execute();没有返回值，因此不用
        ForkJoinTask<Long> submit = forkJoinPool.submit(task);
        Long sum = submit.get();
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"时间"+(end-start));
    }

    // Stream并行流
    public static void test3(){
        long start = System.currentTimeMillis();
        long sum = LongStream.rangeClosed(0L,10_0000_0000L).parallel().reduce(0,Long::sum);
        long end = System.currentTimeMillis();
        System.out.println("sum="+sum+"时间"+(end-start));
    }
}
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

最终结果如下：

可以看出，Stream流计算速度最快！必须要好好掌握Stream流，且虽然看起来ForkJoin相较于for循环差别不大，但是因为ForkJoin的操作非常像递归，有更大的操作空间，如果基线选的好，也许会产生意想不到的效果。

异步回调

package future;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 *
 * 异步调用：Ajax
 */
public class Demo01 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
//        // 没有返回值的runAsync异步回调
//        CompletableFuture completableFuture = CompletableFuture.runAsync(()->{
//            try {
//                TimeUnit.SECONDS.sleep(2);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
//            System.out.println(Thread.currentThread().getName()+"runAsync=>void");
//        });
//
//        System.out.println("1111111");
//        completableFuture.get(); // 获取执行结果

        // 有返回值的异步回调
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName()+"completableFuture=>void");
           // int i = 10/0;// 故意执行错误
            return 1024;
        });
        System.out.println(completableFuture.whenComplete((t, u) -> {
            System.out.println("t---->" + t);// 正常的返回结果
            System.out.println("u---->" + u);// 错误信息
        }).exceptionally((e) -> {
            System.out.println(e.getMessage());
            return 404;
        }).get());
    }
}
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