011Java并发包014异步调用

1 Future

1.1 说明

最开始学习多线程的时候，除了通过继承Thread类创建线程外，还可以通过实现Runnable接口创建线程，并且Thread类本身就是Runnable接口的实现类。

但是Runnable接口有个问题，那就是没有返回值，因此又学习了可以通过实现Callable接口创建线程，使用FutureTask类的构造方法传入Callable接口的实现类，创建可以获取返回值的线程。

Future接口是JDK1.5之后新增的，目的是用于获取异步执行的结果，FutureTask类是RunnableFuture接口的实现类，RunnableFuture接口同时继承了Runnable接口和Future接口，所以使用FutureTask类和Callable接口创建线程，其底层还是通过Runnable接口实现的。

在Executor中提供的execute方法之外，在Executor的子接口ExecutorService中还提供了submit方法，submit方法的返回值是Future类型的对象。

1.2 在线程池中的使用

1.2.1 通过Callable获得返回值

使用Callable作为入参，获取Future结果：

1  Future submit(Callable task);

这种方式是最常用的一种，通过Callable对象存储结果，通过Future对象返回获取结果。

举例如下：

 1 public static void main(String[] args) {
 2     ExecutorService executor = Executors.newSingleThreadExecutor();
 3     Future future = executor.submit(
 4             new Callable() {
 5                 @Override
 6                 public String call() throws Exception {
 7                     return "Hello World";
 8                 }
 9             }
10     );
11     try {
12         System.out.print(future.get());
13     } catch (Exception e) {
14         e.printStackTrace();
15     } finally {
16         executor.shutdown();
17     }
18 }

1.2.2 通过传入指定对象获取返回值

使用Runnable和返回对象作为入参，获取Future结果：

1  Future submit(Runnable task, T result);

通过传入的对象存储返回值，通过Future对象返回获取结果。

举例如下：

 1 public static void main(String[] args) {
 2     final String[] result = new String[1];
 3     ExecutorService executor = Executors.newSingleThreadExecutor();
 4     Future future = executor.submit(
 5             new Runnable() {
 6                 @Override
 7                 public void run() {
 8                     result[0] = "Hello World";
 9                 }
10             }, result
11     );
12     try {
13         System.out.print(future.get()[0]);
14     } catch (Exception e) {
15         e.printStackTrace();
16     } finally {
17         executor.shutdown();
18     }
19 }

1.2.3 无需返回值

使用Runnable作为入参：

1 Future<?> submit(Runnable task);

这种方式无需获取返回值。

举例如下：

 1 public static void main(String[] args) {
 2     ExecutorService executor = Executors.newSingleThreadExecutor();
 3     Future future = executor.submit(
 4             new Runnable() {
 5                 @Override
 6                 public void run() {
 7                 }
 8             }
 9     );
10     try {
11         System.out.print(future.get());
12     } catch (Exception e) {
13         e.printStackTrace();
14     } finally {
15         executor.shutdown();
16     }
17 }

2 CompletableFuture

2.1 说明

虽然Future以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，只能通过阻塞或者轮询的方式得到任务的结果。阻塞的方式显然和异步编程的初衷相违背，轮询的方式又会耗费无谓的CPU资源，而且也不能及时地得到计算结果。

在JDK1.8中，CompletableFuture提供了非常强大的Future的扩展功能，简化了异步编程的复杂性，并且提供了函数式编程的能力，可以通过回调的方式处理计算结果，也提供了转换和组合CompletableFuture的方法。

CompletableFuture实现了CompletionStage接口和Future接口，CompletionStage代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段。一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发。

2.2 使用

2.2.1 新建

CompletableFuture提供了四个静态方法来创建一个异步操作：

1 public static CompletableFuture runAsync(Runnable runnable)
2 public static CompletableFuture runAsync(Runnable runnable, Executor executor)
3 public static  CompletableFuture supplyAsync(Supplier supplier)
4 public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor)

runAsync表示创建无返回值的异步任务，相当于Runnable作为参数的submit方法：

 1 public static void main(String[] args) {
 2     CompletableFuture cf = CompletableFuture.runAsync(()->{
 3         System.out.println(Thread.currentThread().getName() + "-runAsync");
 4     });
 5     try {
 6         System.out.println(Thread.currentThread().getName() + "-result-" + cf.get());
 7     } catch (Exception e) {
 8         e.printStackTrace();
 9     }
10 }

结果如下：

1 ForkJoinPool.commonPool-worker-9-runAsync
2 main-result-null

supplyAsync表示创建带返回值的异步任务的，相当于Callable作为参数的submit方法：

 1 public static void main(String[] args) {
 2     CompletableFuture cf = CompletableFuture.supplyAsync(()->{
 3         System.out.println(Thread.currentThread().getName() + "-supplyAsync");
 4         return "test";
 5     });
 6     try {
 7         System.out.println(Thread.currentThread().getName() + "-result-" + cf.get());
 8     } catch (Exception e) {
 9         e.printStackTrace();
10     }
11 }

结果如下：

1 ForkJoinPool.commonPool-worker-9-supplyAsync
2 main-result-test

2.2.2 完成时回调

当CompletableFuture的计算结果完成，可以执行特定的方法：

1 public CompletableFuture whenComplete(BiConsumer<? super T,? super Throwable> action)
2 public CompletableFuture whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
3 public CompletableFuture whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

whenComplete会在当前执行的线程继续执行CompletableFuture定义的任务：

 1 public static void main(String[] args) {
 2     CompletableFuture future = new CompletableFuture();
 3     new Thread(() -> {
 4         System.out.println(Thread.currentThread().getName() + "-启动");
 5         try {
 6             Thread.sleep(1000);
 7         } catch (Exception e) {
 8             e.printStackTrace();
 9         }
10         System.out.println(Thread.currentThread().getName() + "-完成");
11         future.complete(100);
12         System.out.println(Thread.currentThread().getName() + "-结束");
13     }).start();
14     future.whenComplete((v, ex) -> {
15         try {
16             Thread.sleep(1000);
17         } catch (Exception e) {
18             e.printStackTrace();
19         }
20         System.out.println(Thread.currentThread().getName() + "-v-" + v);
21         System.out.println(Thread.currentThread().getName() + "-ex-" + ex);
22     });
23     try {
24         System.out.println(Thread.currentThread().getName() + "-return-" + future.get());
25     } catch (Exception e) {
26         e.printStackTrace();
27     }
28 }

执行结果：

1 Thread-0-启动
2 //等待1s
3 Thread-0-完成
4 main-return-100
5 //等待1s
6 Thread-0-v-100
7 Thread-0-ex-null
8 Thread-0-结束

whenCompleteAsync会在线程池起一个新的线程执行CompletableFuture定义的任务：

 1 public static void main(String[] args) {
 2     CompletableFuture future = new CompletableFuture();
 3     new Thread(() -> {
 4         System.out.println(Thread.currentThread().getName() + "-启动");
 5         try {
 6             Thread.sleep(1000);
 7         } catch (Exception e) {
 8             e.printStackTrace();
 9         }
10         System.out.println(Thread.currentThread().getName() + "-完成");
11         future.complete(100);
12         System.out.println(Thread.currentThread().getName() + "-结束");
13     }).start();
14     future.whenCompleteAsync((v, ex) -> {
15         try {
16             Thread.sleep(1000);
17         } catch (Exception e) {
18             e.printStackTrace();
19         }
20         System.out.println(Thread.currentThread().getName() + "-v-" + v);
21         System.out.println(Thread.currentThread().getName() + "-ex-" + ex);
22     });
23     try {
24         System.out.println(Thread.currentThread().getName() + "-return-" + future.get());
25         // 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭
26         Thread.sleep(2000);
27     } catch (Exception e) {
28         e.printStackTrace();
29     }
30 }

执行结果：

1 Thread-0-启动
2 //等待1s
3 Thread-0-完成
4 main-return-100
5 Thread-0-结束
6 //等待1s
7 ForkJoinPool.commonPool-worker-9-v-100
8 ForkJoinPool.commonPool-worker-9-ex-null
9 //等待1s

2.2.3 异常处理

当CompletableFuture的计算抛出异常的时候，可以执行特定的方法：

1 public CompletableFuture exceptionally(Functionextends T> fn)

exceptionally会在抛出异常的时候执行：

 1 public static void main(String[] args) {
 2     CompletableFuture future = new CompletableFuture();
 3     new Thread(() -> {
 4         future.completeExceptionally(new NullPointerException());
 5     }).start();
 6     future.exceptionally(e -> {
 7         e.printStackTrace();
 8         return 100;
 9     });
10 }

结果如下：

1 java.lang.NullPointerException

2.2.4 线程依赖

当一个线程依赖另一个线程时，可以使用方法来把这两个线程串行化：

1 public  CompletableFuture thenApply(Function<? super T,? extends U> fn)
2 public  CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn)
3 public  CompletableFuture thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

串行化执行：

 1 public static void main(String[] args) {
 2     CompletableFuture future = CompletableFuture.supplyAsync(()->{
 3         System.out.println(Thread.currentThread().getName() + "-supplyAsync");
 4         return "test";
 5     }).thenApply(t->{
 6         System.out.println(Thread.currentThread().getName() + "-t-" + t);
 7         return t + "-apply";
 8     });
 9     try {
10         System.out.println(Thread.currentThread().getName() + "-" + future.get());
11     } catch (Exception e) {
12         e.printStackTrace();
13     }
14 }

结果如下：

1 ForkJoinPool.commonPool-worker-9-supplyAsync
2 ForkJoinPool.commonPool-worker-9-t-test
3 main-test-apply

2.2.5 消费处理

接收任务的处理结果，并消费处理，无返回结果：

1 public CompletionStage thenAccept(Consumer<? super T> action)
2 public CompletionStage thenAcceptAsync(Consumer<? super T> action)
3 public CompletionStage thenAcceptAsync(Consumer<? super T> action,Executor executor)

消费处理结果：

 1 public static void main(String[] args) {
 2     CompletableFuture future = CompletableFuture.supplyAsync(()->{
 3         System.out.println(Thread.currentThread().getName() + "-supplyAsync");
 4         return "test";
 5     }).thenAccept(t->{
 6         System.out.println(Thread.currentThread().getName() + "-t-" + t);
 7     });
 8     try {
 9         System.out.println(Thread.currentThread().getName() + "-" + future.get());
10     } catch (Exception e) {
11         e.printStackTrace();
12     }
13 }

结果如下：

1 ForkJoinPool.commonPool-worker-9-supplyAsync
2 main-t-test
3 main-null

2.2.6 取消返回

不接收任务的处理结果，取消返回：

1 public CompletionStage thenRun(Runnable action)
2 public CompletionStage thenRunAsync(Runnable action)
3 public CompletionStage thenRunAsync(Runnable action,Executor executor)

取消返回：

 1 public static void main(String[] args) {
 2     CompletableFuture future = CompletableFuture.supplyAsync(()->{
 3         System.out.println(Thread.currentThread().getName() + "-supplyAsync");
 4         return "test";
 5     }).thenRun(()->{
 6         System.out.println(Thread.currentThread().getName() + "-t-null");
 7     });
 8     try {
 9         System.out.println(Thread.currentThread().getName() + "-" + future.get());
10     } catch (Exception e) {
11         e.printStackTrace();
12     }
13 }

结果如下：

1 ForkJoinPool.commonPool-worker-9-supplyAsync
2 main-t-null
3 main-null
javaJava并发包

相关

JavaWeb的三大作用域

java中常用数组copy方法

6.Java方法

Java web开发：从零到 Run 一个现有的 Spring Boot 前后端分离项目（前端Vue）

javascript基础知识(29) 实现继承的方式

java 正则表达式

JAVA List<Object[]>取值问题

初次接触Java感受

99乘法表 java for循环

基于java.net.socket包的对象传递

Idea Java项目文件夹后面提示百分数怎么去掉

Java8新特性

标签

一品网 冀ICP备14022925号-6