JUC并发编程之CompletableFuture
Future
future是java5新加的一个接口，他提供了一种异步并行计算的功能
接口定义了操作异步任务执行的一些方法，如获取异步任务的执行结果、取消任务的执行、判断任务是否被取消、判断任务是否执行完毕
目的：异步多线程执行且有返回结果，特点：多线程/有返回/异步任务
补充：Runnable实现的是run方法，没有返回值，没有异常，Callable实现的是call方法，有返回值，需要处理异常
Future接口常用实现类Future Task异步任务
代码实现
public class CompletableFutureDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {FutureTask<String> futureTask = new FutureTask<>(new MyThread());Thread t1 = new Thread(futureTask,"t1");t1.start();System.out.println(futureTask.get());} }class MyThread implements Callable<String>{@Overridepublic String call() throws Exception {System.out.println("----come in call()");return "hello Callable";} }
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、获得结果和触发计算
public class CompletableFutureAPIDemo {public static void main(String[] args) {CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}return "abc";});//System.out.println(completableFuture.get()); 阻塞主线程获取结果//System.out.println(completableFuture.get(2L,TimeUnit.SECONDS)); 等待指定时间获取结果//System.out.println(completableFuture.join()); 阻塞线程获取结果System.out.println(completableFuture.getNow("xxx")); //立即获取结果不阻塞，如果没计算完成返回XXXSystem.out.println(completableFuture.complete("completeValue")+"\\t"+completableFuture.join());//是否打断get方法立刻获得括号值}
}2、对计算结果进行处理
public class CompletableFutureAPI2Demo {public static void main(String[] args) {ExecutorService threadPool = Executors.newFixedThreadPool(3);//thenApply//计算结果存在依赖关系，这两个线程串行化//异常相关：由于存在依赖关系（当前步错，不走下一步），当前步骤有异常的话就叫停//handle:区别：有异常也可以往下一步走，根据带的异常参数可以进一步处理CompletableFuture.supplyAsync(() -> {try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("111");return 1;},threadPool).handle((f,e) -> {int i = 10/0;System.out.println("222");return f+2;}).thenApply(f -> {System.out.println("333");return f+3;}).whenComplete((v,e) -> {if (e == null){System.out.println("------计算结果:"+v);}}).exceptionally(e -> {e.printStackTrace();System.out.println("e = " + e.getMessage());return null;});threadPool.shutdown();System.out.println(Thread.currentThread().getName()+"-----主线程先去忙其他任务");}
}3、对计算结果进行消费
//接受任务的处理结果，并消费处理，无返回结果
public class CompletableFutureAPI3Demo {public static void main(String[] args) {
// CompletableFuture.supplyAsync(() -> {
// return 1;
// }).thenApply(f -> {
// return f+2;
// }).thenApply(f -> {
// return f+3;
// }).thenAccept(System.out::println);//任务之间的顺序执行/* 1、thenRun：任务A执行完执行任务B，并且B不需要A的结果* 2、thenAccept：任务 A执行完执行B,B需要A的结果，但是任务B无返回值* 3、thenApply：任务A执行完执行B，B需要A的结果，同时任务B有返回值*/System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {}).join());System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenAccept(r-> System.out.println("r = " + r)).join());System.out.println(CompletableFuture.supplyAsync(() -> "resultA").thenApply(r -> r+"ResuleB").join());}
}4、对计算速度进行选用
public class CompletableFutureFastDemo {public static void main(String[] args) {CompletableFuture<String> playA = CompletableFuture.supplyAsync(() -> {System.out.println("A come in");try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}return "playA";});CompletableFuture<String> playB = CompletableFuture.supplyAsync(() -> {System.out.println("B come in");try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}return "playB";});CompletableFuture<String> result = playA.applyToEither(playB, f -> f + " is winer");System.out.println(Thread.currentThread().getName()+"\\t"+"--------"+result.join());}
}5、对计算结果进行合并
public class CompletableFutureCombineDemo {public static void main(String[] args) {CompletableFuture<Integer> completableFuture1 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\\t ---启动");try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}return 10;});CompletableFuture<Integer> completableFuture2 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\\t ---启动");try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {e.printStackTrace();}return 20;});CompletableFuture<Integer> result = completableFuture1.thenCombine(completableFuture2, (x, y) -> {System.out.println("开始两个结果合并:" + x + y);return x + y;});System.out.println("result.join() = " + result.join());}
}注：CompletableFuture线程池运行选择
1、没有传入自定义线程池，都用默认线程池ForkJoinPool
2、传入了一个自定义线程池
如果你执行第一个任务的时候，传入了一个自定义线程池
调用thenRun方法执行第二个任务时，第二个任务和第一个任务使用的是同一个线程池
调用thenRunAsync方法执行第二个任务时，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池
备注：
有可能处理太快，系统优化切换原则，直接使用main线程处理
public class CompletableFutureWithThreadPoolDemo {public static void main(String[] args) {ExecutorService threadPool = Executors.newFixedThreadPool(5);try {CompletableFuture<Void> completableFuture = CompletableFuture.supplyAsync(() -> {try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("1号任务" + "\\t" + Thread.currentThread().getName());return "abcd";},threadPool).thenRunAsync(() -> {try {TimeUnit.MILLISECONDS.sleep(20);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("2号任务" + "\\t" + Thread.currentThread().getName());}).thenRun(() -> {try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("3号任务" + "\\t" + Thread.currentThread().getName());}).thenRun(() -> {try {TimeUnit.MILLISECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("4号任务" + "\\t" + Thread.currentThread().getName());});System.out.println(completableFuture.get(2L, TimeUnit.SECONDS));} catch (Exception e) {e.printStackTrace();}finally {threadPool.shutdown();}}
}
