JUC-05-ForkJoin，Future（了解即可）

文章目录

01、ForkJoin：分支合并
02、java.util.concurrent.Future接口

01、ForkJoin：分支合并

java.util.concurrent包下的实现类：

java.util.concurrent.ForkJoinTask
java.util.concurrent.ForkJoinPool

ForkJoin特点：工作窃取

ForkJoin中维护的是双端队列，这样才可以工作窃取，当一个执行快的线程把本队列中的任务做完，去帮助别的线程执行任务，这时两个线程一个从队尾执行，一个从对头执行（类似于执行快的线程偷偷执行了执行慢的线程的任务）

ForkJoin使用方法：

1、创建一个自己计算任务类，继承ForkJoinTask，里面是一个任务ForkJoinTask
2、创建ForkJoinPool的实现类，通过它里面的execute(ForkJoinTask task)方法来执行ForkJoinTask
3，通过ForkJoinPool中的submit(ForkJoinTask<T> task)来提交ForkJoinTask任务并执行

注意：

ForkJoinPool是一个普通类，ForkJoinTask是抽象类
我们一般继承ForkJoinTask的子类（子类也是抽象类）：-  需要返回值继承：RecursiveTask- 不需要返回值继承：RecursiveAction

用一个计算任务为例（难度依次增加）：

- 1.普通方法计算
- 2.使用ForkJoin

public class ForkJoinTest{public static void main(String[] args)throws Exception {//test01();test02();//test3();}//普通方法计算public static void test01(){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 test02() throws ExecutionException, InterruptedException {//开始时间long start = System.currentTimeMillis();//2、创建ForkJoinPool的实例，是用来执行ForkJoinTaskForkJoinPool forkJoinPool = new ForkJoinPool();//1、创建ForkJoinTaskForkJoinTask<Long> task = new ForkJoinDemo(0L, 10_0000_0000L);//2、提交任务ForkJoinTask<Long> submit = forkJoinPool.submit(task);//获得结果Long sum = submit.get();long end = System.currentTimeMillis();System.out.println("sum="+sum+" 时间："+(end-start));}public static void test3(){long start = System.currentTimeMillis();// Stream并行流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、创建一个类继承ForkJoin的子类RecursiveTask，也是抽象类
class ForkJoinDemo extends RecursiveTask<Long> {private Long start;private Long end;private Long temp = 10000L;// 临界值public ForkJoinDemo(Long start, Long end) {this.start = start;this.end = end;}//计算方法@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, end);task2.fork(); // 拆分任务，把任务压入线程队列return task1.join() + task2.join();//合并任务并返回}}
}

02、java.util.concurrent.Future接口

Future接口是Java标准API的一部分，在java.util.concurrent包中，该接口下的类可以用来异步执行任务。

Future模式可以这样来描述：

我有一个任务，提交给了Future，Future替我完成这个任务。期间我自己可以去做任何想做的事情。一段时间之后，我就便可以从Future那儿取出结果。
就相当于下了一张订货单，一段时间后可以拿着提订单来提货，这期间可以干别的任何事情，其中Future 接口就是订货单，真正处理订单的是Executor类，它根据Future接口的要求来生产产品。

Future接口提供方法来检测任务是否被执行完，等待任务执行完获得结果，也可以设置任务执行的超时时间。这个设置超时的方法就是实现Java程序执行超时的关键。

Future接口是一个泛型接口，严格的格式应该是Future<V>，其中V代表了Future执行的任务返回值的类型。

Future接口的方法介绍如下：

boolean cancel (boolean mayInterruptIfRunning) ：取消任务的执行。参数指定是否立即中断任务执行，或者等任务结束
boolean isCancelled () ：任务是否已经取消，任务正常完成前将其取消，则返回 true
boolean isDone () ：任务是否已经完成，需要注意的是如果任务正常终止、异常或取消，都将返回true
V get () throws InterruptedException, ExecutionException 等待任务执行结束，然后获得V类型的结果。InterruptedException 线程被中断异常， ExecutionException任务执行异常，如果任务被取消，还会抛出CancellationException
V get (long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 同上面的get功能一样，多了设置超时时间，参数timeout指定超时时间，uint指定时间的单位，在枚举类TimeUnit中有相关的定义。如果计算超时，将抛出TimeoutException。

//使用Executor创建一个线程
ExecutorService executor = Executors.newSingleThreadExecutor();
//创建一个FutureTask的实例
FutureTask<String> future =  new FutureTask<String>(new Callable<String>() {//使用Callable接口作为构造参数  public String call() {  //真正的任务在这里执行，这里的返回值类型为String，可以为任意类型  }});
//提交执行这个任务future，异步任务
executor.execute(future);
//在这里可以做别的任何事情
try {  result = future.get(5000, TimeUnit.MILLISECONDS); //取得结果，同时设置超时执行时间为5秒。同样可以用future.get()，不设置执行超时时间取得结果
} catch (InterruptedException e) {  futureTask.cancel(true);
} catch (ExecutionException e) {  futureTask.cancel(true);
} catch (TimeoutException e) {  futureTask.cancel(true);
} finally {  executor.shutdown();
}

Future接口实现类：都可以异步执行任务

- CompletableFuture
- ForkJoinTask
- FutureTask- 通常使用FutureTask来处理我们的任务。- FutureTask类同时又实现了Runnable接口，所以可以直接提交给Executor执行。

使用CompletableFuture进行异步调用步骤：

- 异步执行
- 成功回调
- 失败回调

CompletableFuture是JDK1.8版本新引入的类，这个类实现了俩接口Future和CompletionStage：

- 使用CompletionStage接口，去支持完成时触发的函数和操作，能做一些ExecutorService配合Future做不了的工作
- ExecutorService配合Future需要等待isDone为true才能知道任务跑完了或者就是用get方法调用的时候会出现阻塞
- 而CompletableFuture就可以用then，when等等操作来防止以上的阻塞和轮询isDone的现象出现， 【也可以使用get方法调用的时候会出现阻塞】

一个CompletableFuture对象代表着一个任务：

- 创建CompletableFuture直接new对象也成
- complete意思就是这个任务完成了需要返回的结果
- 然后用get();方法可以获取到

public class Demo01 {public static void main(String[] args) throws InterruptedException,ExecutionException {runAsyncVoid();supplyAsyncReturn();}public static void runAsyncVoid()throws InterruptedException,ExecutionException{// 没有返回值的 runAsync 异步回调，【Async：异步】CompletableFuture<Void> completableFuture=CompletableFuture.runAsync(()->{try {//当线程沉睡时不阻塞，其他线程可执行TimeUnit.SECONDS.sleep(6);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName()+"runAsync=>Void");});System.out.println("666666");completableFuture.get();//阻塞获取结果，在获取结果之前后面的代码不执行System.out.println("88888888");}public static void supplyAsyncReturn() throws InterruptedException,ExecutionException{//有返回值的supplyAsync，【async：异步】CompletableFuture<Integer> completableFuture=CompletableFuture.supplyAsync(()->{System.out.println(Thread.currentThread().getName()+"supplyAsync=>Integer");int i = 10/0;return 1024;});//System.out.println("22222222");//completableFuture.get();//获取返回结果，不会阻塞，但是不处理错误，遇到错误会停止执行//System.out.println("33333333");//whenComplete中参数是 BiConsumer，有两个返回值，第一个是结果，第二个是错误信息//因为BiConsumer是消费型接口，无法返回错误信息，要想返回错误信，需要使用exceptionally//链式编程：返回值都一样，可以像链子一样，根据返回值再次调用System.out.println(completableFuture.whenComplete((t, u) -> {System.out.println("t=>" + t); // 正常的返回结果System.out.println("u=>" + u); // 错误信息：//java.util.concurrent.CompletionException: java.lang.ArithmeticException:by zero}).exceptionally((e) -> {System.out.println(e.getMessage());return 233; // 可以获取到错误的返回结果}).get());}
}

运行结果

666666
ForkJoinPool.commonPool-worker-9runAsync=>Void
88888888
ForkJoinPool.commonPool-worker-9supplyAsync=>Integer
t=>null
u=>java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
java.lang.ArithmeticException: / by zero
233

JUC-05-ForkJoin，Future（了解即可）相关推荐

【重难点】【JUC 05】线程池核心设计与实现、线程池使用了什么设计模式、要你设计的话，如何实现一个线程池
[重难点][JUC 05]线程池核心设计与实现.线程池使用了什么设计模式.要你设计的话,如何实现一个线程池文章目录 [重难点][JUC 05]线程池核心设计与实现.线程池使用了什么设计模式.要你设计 ...
Java JUC工具类--Future
Future模式,也是非常经典的设计模式,这种模式主要就利用空间换时间的概念,也就是说异步执行(需要开启一个新的线程) 在互联网高并发的应用服务中,我们随处可见这种理念和代码,主要就是使用了这种模式 ...
Java——聊聊JUC中的Future和FutureTask
1.回顾Future和FutureTask 首先,Future是一个接口(Java5就诞生了),而FutureTask是它的实现类.而Future接口则是定义了操作异步任务执行的一些方法,提供了一种异 ...
【JUC系列】Future异步回调模式
何为异步回调前面只是一个例子,对并发的主要模式进行形象的说明. 下面正式来讲下经常使用的几个和并发相关的概念. 1.2.1. 同步.异步.阻塞.非阻塞一:同步所谓同步,就是在发出一个功能调用时, ...
JUC之ForkJoin框架
ForkJoin ForkJoin是由JDK1.7后提供多线并发处理框架, ForkJoin的框架的基本思想是分而治之.使用ForkJoin将相同的计算任务通过多线程的进行执行, 从而能提高数据的计算 ...
Java——聊聊JUC中的CompletableFuture
文章目录: 1.承接Future和FutureTask 2.CompletableFuture四大静态方法 2.1 runAsync(Runnable runnable) 2.2 runAsync(R ...
java 检视_Java高并发系列——检视阅读（五）
JUC中工具类CompletableFuture CompletableFuture是java8中新增的一个类,算是对Future的一种增强,用起来很方便,也是会经常用到的一个工具类,熟悉一下. Co ...
【华为云技术分享】Reactive模式优势与实践
Reactive编程即反应式编程,随着这些年的发展已经逐步的进入了开发者的视野当中.早在2014年社区就有人发起响应式宣言,推动着Reactive的发展: 响应式宣言 Published on Sep ...
Java 并发之 FutureTask 的基本使用
为什么80%的码农都做不了架构师?>>> 上次我们说到了 JUC 中的 Future 接口,在最后提到了 FutureTask.CompletionService 等.我们这次 ...
Netty入门-第二话
文章目录 Netty 概述原生 NIO 存在的问题 Netty 官网说明 Netty 的优点 Netty 版本说明 Netty 高性能架构设计线程模型基本介绍传统阻塞 I/O 服务模型工作原理 ...

JUC-05-ForkJoin，Future（了解即可）

文章目录

01、ForkJoin：分支合并

02、java.util.concurrent.Future接口

JUC-05-ForkJoin，Future（了解即可）相关推荐

最新文章

热门文章