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

本文主要是介绍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接口中唯一抽象方法@Overrideprotected 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();}}
}

测试类：该类中分别用了三种方法去计算求和，普通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));}// forkJoinpublic 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));}
}

最终结果如下：

外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

可以看出，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<Void> 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());}
}