JUC并发编程之CompletableFuture-阿里云开发者社区

Future

future是java5新加的一个接口，他提供了一种异步并行计算的功能

接口定义了操作异步任务执行的一些方法，如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务是否执行完毕

目的：异步多线程执行且有返回结果，特点：多线程/有返回/异步任务

补充：Runnable实现的是run方法，没有返回值，没有异常，Callable实现的是call方法，有返回值，需要处理异常

Future接口常用实现类Future Task异步任务

代码实现

1. public class CompletableFutureDemo {
2. public static void main(String[] args) throws ExecutionException, InterruptedException {
3.         FutureTask<String> futureTask = new FutureTask<>(new MyThread());
4. Thread t1 = new Thread(futureTask,"t1");
5.         t1.start();
6.         System.out.println(futureTask.get());
7.     }
8. }
9. 
10. 
11. 
12. class MyThread implements Callable<String>{
13. 
14. @Override
15. public String call() throws Exception {
16.         System.out.println("----come in call()");
17. return "hello Callable";
18.     }
19. }

future的优缺点

NO: 1、get方法容易阻塞：一旦调用，如果计算没有完成，容易程序阻塞

2、isDone轮询：轮询的方式会耗费CPU资源，而且也不见得能及时得到计算结果

小结：future对于结果的获取不是很友好，只能通过阻塞或轮询的方式得到任务结果

注：实际程序中轮询的方式并不比阻塞好，这么写是给用户看的，用户不可能看你的程序就停在那了，这么写可以给个提示，也不会直接异常

CompletableFuture的异步优化思路

背景：对于简单的业务场景使用Future完全OK，但是对于复杂的业务场景，比如：

1、回调通知：应对Future的完成时间，完成了可以告诉我，也就是我们的回调通知，通过轮询的方式去判断任务是否完成这样非常的占用CPU并且代码也不优雅

2、创建异步任务：Future+线程池配合

3、多个任务前后依赖可以组合处理：想将多个异步任务的计算结果组合起来，后一个异步任务的计算结果需要前一个异步任务结果的值，将两个或多个异步计算合成一个异步计算，这几个异步计算相互独立，同时后面这个又依赖前一个处理的结果

4、对计算速度最快：当Future集合中某个任务最快结束时，返回结果，返回第一名处理结果

CompletableFuture提供了一种观察者模式类似的机制，可以让任务完成后通知监听的一方

类架构说明

接口CompletionStage

1、CompletionStage代表异步计算过程中某一个阶段，一个阶段完成以后可能会触发另外一个阶段

2、一个阶段的执行可以是一个Function，Consumer或者Runnable

3、一个阶段的执行可能是被单个阶段的完成触发，也可能是由多个阶段一起触发

代表异步计算过程中的某一个阶段，一个阶段完成以后可能会触发另外一个阶段

类CompletableFuture

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

2、他可能代表一个明确完成的Future，也有可能代表一个完成阶段，它支持在计算完成以后触发一些函数或执行某个动作

3、它实现了Future和CompletionStage接口

注：从java8开始引入了CompletableFuture，他是Future的功能增强版，减少阻塞和轮询，可以传入回调对象，当异步任务完成或者发生异常时，自动调用回调对象的回调方法

优点：

1、异步任务结束时，会自动回调某个对象的方法

2、主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行

3、异步任务出错时，会自动回调某个对象的方法

函数式接口复习

CompletableFuture常用方法

1、获得结果和触发计算

1. public class CompletableFutureAPIDemo {
2. public static void main(String[] args) {
3.         CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
4. try {
5.                 TimeUnit.SECONDS.sleep(1);
6.             } catch (InterruptedException e) {
7.                 e.printStackTrace();
8.             }
9. return "abc";
10.         });
11. 
12. //System.out.println(completableFuture.get());   阻塞主线程获取结果
13. //System.out.println(completableFuture.get(2L,TimeUnit.SECONDS)); 等待指定时间获取结果
14. //System.out.println(completableFuture.join());  阻塞线程获取结果
15.         System.out.println(completableFuture.getNow("xxx")); //立即获取结果不阻塞，如果没计算完成返回XXX
16.         System.out.println(completableFuture.complete("completeValue")+"\t"+completableFuture.join());
17. //是否打断get方法立刻获得括号值
18.     }
19. }

2、对计算结果进行处理

1. public class CompletableFutureAPI2Demo {
2. public static void main(String[] args) {
3. 
4. ExecutorService threadPool = Executors.newFixedThreadPool(3);
5. 
6. //thenApply
7. //计算结果存在依赖关系，这两个线程串行化
8. //异常相关：由于存在依赖关系（当前步错，不走下一步），当前步骤有异常的话就叫停
9. //handle:区别：有异常也可以往下一步走，根据带的异常参数可以进一步处理
10.         CompletableFuture.supplyAsync(() -> {
11. try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}
12.             System.out.println("111");
13. return 1;
14.         },threadPool).handle((f,e) -> {
15. int i = 10/0;
16.             System.out.println("222");
17. return f+2;
18.         }).thenApply(f -> {
19.             System.out.println("333");
20. return f+3;
21.         }).whenComplete((v,e) -> {
22. if (e == null){
23.                 System.out.println("------计算结果:"+v);
24.             }
25.         }).exceptionally(e -> {
26.             e.printStackTrace();
27.             System.out.println("e = " + e.getMessage());
28. return null;
29.         });
30.         threadPool.shutdown();
31.         System.out.println(Thread.currentThread().getName()+"-----主线程先去忙其他任务");
32.     }
33. }

3、对计算结果进行消费

1. //接受任务的处理结果，并消费处理，无返回结果
2. public class CompletableFutureAPI3Demo {
3. public static void main(String[] args) {
4. //        CompletableFuture.supplyAsync(() -> {
5. //            return 1;
6. //        }).thenApply(f -> {
7. //            return f+2;
8. //        }).thenApply(f -> {
9. //            return f+3;
10. //        }).thenAccept(System.out::println);
11. 
12. //任务之间的顺序执行
13. /**
14.          * 1、thenRun：任务A执行完执行任务B，并且B不需要A的结果
15.          * 2、thenAccept：任务  A执行完执行B,B需要A的结果，但是任务B无返回值
16.          * 3、thenApply：任务A执行完执行B，B需要A的结果，同时任务B有返回值
17.          */
18. 
19.         System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {}).join());
20.         System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenAccept(r-> System.out.println("r = " + r)).join());
21.         System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenApply(r -> r+"ResuleB").join());
22. 
23.     }
24. }

4、对计算速度进行选用

1. public class CompletableFutureFastDemo {
2. public static void main(String[] args) {
3.         CompletableFuture<String> playA = CompletableFuture.supplyAsync(() -> {
4.             System.out.println("A come in");
5. try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}
6. return "playA";
7.         });
8. 
9.         CompletableFuture<String> playB = CompletableFuture.supplyAsync(() -> {
10.             System.out.println("B come in");
11. try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}
12. return "playB";
13.         });
14.         CompletableFuture<String> result = playA.applyToEither(playB, f -> f  + " is winer");
15.         System.out.println(Thread.currentThread().getName()+"\t"+"--------"+result.join());
16.     }
17. }

5、对计算结果进行合并

1. public class CompletableFutureCombineDemo {
2. public static void main(String[] args) {
3.         CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {
4.             System.out.println(Thread.currentThread().getName() + "\t ---启动");
5. try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}
6. return 10;
7.         });
8. 
9.         CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {
10.             System.out.println(Thread.currentThread().getName() + "\t ---启动");
11. try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}
12. return 20;
13.         });
14. 
15.         CompletableFuture<Integer> result = completableFuture1.thenCombine(completableFuture2, (x, y) -> {
16.             System.out.println("开始两个结果合并:" + x + y);
17. return x + y;
18.         });
19.         System.out.println("result.join() = " + result.join());
20.     }
21. }

注：CompletableFuture线程池运行选择

1、没有传入自定义线程池，都用默认线程池ForkJoinPool

2、传入了一个自定义线程池

如果你执行第一个任务的时候，传入了一个自定义线程池

调用thenRun方法执行第二个任务时，第二个任务和第一个任务使用的是同一个线程池

调用thenRunAsync方法执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池

备注：

有可能处理太快，系统优化切换原则，直接使用main线程处理

1. public class CompletableFutureWithThreadPoolDemo {
2. public static void main(String[] args) {
3. ExecutorService threadPool = Executors.newFixedThreadPool(5);
4. 
5. try {
6.             CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {
7. try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
8.                 System.out.println("1号任务" + "\t" + Thread.currentThread().getName());
9. return "abcd";
10.             },threadPool).thenRunAsync(() -> {
11. try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}
12.                 System.out.println("2号任务" + "\t" + Thread.currentThread().getName());
13.             }).thenRun(() -> {
14. try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}
15.                 System.out.println("3号任务" + "\t" + Thread.currentThread().getName());
16.             }).thenRun(() -> {
17. try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}
18.                 System.out.println("4号任务" + "\t" + Thread.currentThread().getName());
19.             });
20.             System.out.println(completableFuture.get(2L, TimeUnit.SECONDS));
21.         } catch (Exception e) {
22.             e.printStackTrace();
23.         }finally {
24.             threadPool.shutdown();
25.         }
26.     }
27. }

JUC并发编程之CompletableFuture

Future

CompletableFuture的异步优化思路

CompletableFuture常用方法

1、获得结果和触发计算

2、对计算结果进行处理

3、对计算结果进行消费

4、对计算速度进行选用

5、对计算结果进行合并

热门文章

最新文章

相关电子书