CompletableFuture使用教学
一、开始一个线程异步执行不需要返回值
通过runAsync方式
//1.不加线程池方式CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {System.out.println(Thread.currentThread().getName());//停顿几秒try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}});System.out.println("hello world");System.out.println(completableFuture.get());//null 没有返回值的情况//2.加线程池方式//创建固定线程池（阿里规范建议使用自定义线程池，不能通过Executors来进行创建）ExecutorService executors = Executors.newFixedThreadPool(4);//此处偷懒，用此线程池CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {System.out.println(Thread.currentThread().getName());//停顿几秒try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}}, executors);System.out.println(runAsync.get());二、通过异步方式执行，有返回值
supplyAsync
//不加线程池方式CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName());try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}return "hello supplyAsync";});System.out.println(completableFuture.get());//加线程池的方式ExecutorService executors = Executors.newFixedThreadPool(4);CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName());try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}return "hello supplyAsync+Executors";}, executors);System.out.println(completableFuture.get());ps：get（）方法可以获取异步线程执行完后的结果
三.通过whenComplete减少阻塞和轮询(可加线程池,也可不加)
即当异步线程执行结束会接着执行whenComplete（）方法，如果执行期间报错会执行exceptionally()方法ExecutorService threadPool = Executors.newFixedThreadPool(6);CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "--副线程");int result = new Random().nextInt(10);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println("1秒后出结果");return result;},threadPool).whenComplete((v,e)->{//没有异常v是值 e是异常情况if (e == null){System.out.println("计算完成,UpdateValue:"+v);}}).exceptionally((e)->{//e是异常情况e.printStackTrace();System.out.println("异常情况:"+e.getCause()+"\t"+e.getMessage());return null;});四、实现通过货品在不同平台进行价格搜索进行数据汇总
普通方式实现
public class Case {static List<NetMall> list = Arrays.asList(new NetMall("jd"),new NetMall("dangdang"),new NetMall("taobao"));public static List<String> getPrice(List<NetMall> list,String productName){return list.stream().map(netMall -> String.format(productName+ " in %s price is %s",netMall.getNetMallName(),netMall.calcPrice(productName))).collect(Collectors.toList());}public static void main(String[] args) {long start = System.currentTimeMillis();List<String> list1 = getPrice(list, "mysql");for (String s : list1) {System.out.println(s);}long end = System.currentTimeMillis();System.out.println("---当前操作时间--costTime:"+(end - start)+"ms");}}class NetMall{private String netMallName;public String getNetMallName() {return netMallName;}public NetMall(){}public NetMall(String netMallName){this.netMallName = netMallName;}public double calcPrice(String productName){try {TimeUnit.SECONDS.sleep(1);//此处表示 业务执行所需耗时时间}catch (InterruptedException e){e.printStackTrace();}return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);//模拟价格}}普通方式实现耗时结果：
completableFuture实现
public class Case {static List<NetMall> list = Arrays.asList(new NetMall("jd"),new NetMall("dangdang"),new NetMall("taobao"));public static List<String> getPricesByCompletableFuture(List<NetMall> list,String productName){return list.stream().map(netMall -> CompletableFuture.supplyAsync(()->String.format(productName+"in %s price is %.2f",netMall.getNetMallName(),netMall.calcPrice(productName)))).collect(Collectors.toList()).stream().map(s->s.join()).collect(Collectors.toList());}public static void main(String[] args) {long start = System.currentTimeMillis();List<String> list1 = getPricesByCompletableFuture(Case.list, "mysql");for (String s : list1) {System.out.println(s);}long end = System.currentTimeMillis();System.out.println("---当前操作时间--costTime:"+(end - start)+"ms");}}class NetMall{private String netMallName;public String getNetMallName() {return netMallName;}public NetMall(){}public NetMall(String netMallName){this.netMallName = netMallName;}public double calcPrice(String productName){try {TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);//模拟价格}}结果耗时：
相比较能够发现，同时开启三个异步线程，时间仅仅为单个平台查询的时间，大大节省效率！
五、CompletableFuture常用API
1.获得结果和触发计算
public T get() 不见不散，容易阻塞public T get(long timeout,TimeUnit unit) 过时不候，超过时间会爆异常public T join() 类似于get()，区别在于是否需要抛出异常public T getNow(T valueIfAbsent)立即获取结果不阻塞计算完，返回计算完成后的结果没算完，返回设定的valueAbsent(直接返回了备胎值xxx)主动触发计算public boolean complete(T value) 是否立即打断get()方法返回括号值（下面代码实现）public static void main(String[] args) throws InterruptedException, ExecutionException {CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {throw new RuntimeException(e);}return "abc";});TimeUnit.SECONDS.sleep(1);System.out.println(completableFuture.complete("end")+"\t"+completableFuture.get());}执行结果：true end
解释：执行需要2秒，等待1秒；
complete（默认值）方法会打断执行，如果执行完，则返回结果，如果没有执行完则输出默认值；
2.对计算结果进行处理
（1）thenApply 计算结果存在在依赖关系，使得线程串行化。因为依赖关系，所以一旦有异常，直接叫停。public static void main(String[] args) {CompletableFuture.supplyAsync(()->{try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("111");return 1024;}).thenApply(f->{System.out.println("222");return f + 1;}).thenApply(f->{System.out.println("333");return f + 1;//return f/0; 会报出异常}).whenCompleteAsync((v,e)->{System.out.println("****v="+v);}).exceptionally(e->{e.printStackTrace();return null;});System.out.println("----主线程结束--end");// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }}（2）handle 类似于thenApply，但是有异常的话仍然可以往下走一步。CompletableFuture.supplyAsync(() -> {//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }System.out.println("111");return 1024;}).handle((f,e) -> {int age = 10/0;//异常语句System.out.println("222");return f + 1;}).handle((f,e) -> {System.out.println("333");return f + 1;}).whenCompleteAsync((v,e) -> {System.out.println("*****v: "+v);}).exceptionally(e -> {e.printStackTrace();return null;});System.out.println("----主线程结束--end");//在222方法中报错 会继续执行输出333（3）.对计算结果进行消费接收任务的处理结果，并消费处理，无返回结果|消费型函数式接口，之前的是FunctionCompletableFuture.supplyAsync(()->{return 1;}).thenApply(f -> {return f+1;}).thenApply(f ->{return f+3;}).thenApply(f->{return f+4;}).thenAccept(r->{System.out.println("r==\t"+r);});补充：Code之任务之间的顺序执行
thenRun
- thenRun(Runnable runnable)
- 任务A执行完执行B，并且B不需要A的结果
thenAccept
- thenAccept(Consumer action)
- 任务A执行完执行B，B需要A的结果，但是任务B无返回值
thenApply
thenApply(Function fn)
任务A执行完执行B，B需要A的结果，同时任务B有返回值
（4）.CompleteFuture和线程池说明（非常重要）
上面的几个方法都有普通版本和后面加Async的版本
以thenRun和thenRunAsync为例，有什么区别？先看结论
1.没有传入自定义线程池，都用默认线程池ForkJoinPool
2.传入了一个自定义线程池如果你执行第一个任务的时候，传入了一个自定义线程池
- 调用thenRun方法执行第二个任务的时候，则第二个任务和第一个任务是用同一个线程池
- 调用thenRunAsync执行第二个任务的时候，则第一个任务使用的是你自己传入的线程池，第二个任务使用的是ForkJoin线程池
3.也有可能处理太快，系统优化切换原则，直接使用main线程处理（把sleep去掉）
（5）.对计算速度进行选用
//applyToEither 线程先执行完的输出，输出其中一个CompletableFuture<String> play1 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }return "play1 ";});CompletableFuture<String> play2 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }return "play2 ";});CompletableFuture<String> thenCombineResult= play1.applyToEither(play2, f -> {return f + "is winner";});System.out.println(Thread.currentThread().getName()+"\t"+thenCombineResult.get());（6）.对计算结果进行合并‘
thenCombine 合并
- 两个CompletionStage任务都完成后，最终能把两个任务的结果一起交给thenCOmbine来处理
- 先完成的先等着，等待其它分支任务
CompletableFuture<Integer> completeFuture1 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in");return 10;});CompletableFuture<Integer> completeFuture2 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in");return 20;});CompletableFuture<Integer> completableFuture = completeFuture1.thenCombine(completeFuture2,(x, y) -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in");return x + y;});System.out.println(completableFuture.get()); //输出 30合并版本
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in1");return 10;}).thenCombine(CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in2");return 20;}), (x, y) -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in3");return x + y;}).thenCombine(CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in4");return 30;}), (x, y) -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in5");return x + y;});System.out.println(completableFuture.get());//输出结果//ForkJoinPool.commonPool-worker-9---come in1//ForkJoinPool.commonPool-worker-9---come in2//main---come in3//ForkJoinPool.commonPool-worker-2---come in4//main---come in5//60