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();}}
}