JUC高级: CompletableFuture

代码实例:

package site.zhourui.juc;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService executorService = Executors.newFixedThreadPool(3);//加入线程池CompletableFuture completableFuture = CompletableFuture.supplyAsync(()->{System.out.println(Thread.currentThread().getName());try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}return "helllo supplyasync";});System.out.println(completableFuture.get());}
}

执行结果:

2.3.3 CompletableFuture通用演示

注意:默认线程池ForkJoin会在主线程执行完成时关闭,如果有任务正在使用该线程池那么可能不会出结果

CompletableFuture完成与Future相同的功能

package site.zhourui.juc;import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;public class CompletableFutureUseDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {future1();}private static void future1() throws ExecutionException, InterruptedException {CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "...come in");int result = ThreadLocalRandom.current().nextInt(10);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("1秒钟后出结果:" + result);return result;});System.out.println(Thread.currentThread().getName()+"线程先去忙其他任务!");System.out.println(completableFuture.get());}
}

执行结果:和FutureTask一致

CompletableFuture通用演示

package site.zhourui.juc;import java.util.concurrent.*;public class CompletableFutureUseDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException {//自定义线程池ExecutorService threadPool = Executors.newFixedThreadPool(3);try {CompletableFuture.supplyAsync(() ->{System.out.println(Thread.currentThread().getName()+"...come in");int result = ThreadLocalRandom.current().nextInt(10);try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("1秒钟后出结果:"+result);if(result > 5){int i = 10/0; //制造异常}return result;},threadPool).whenComplete((v,e) ->{ //v表示result,e表示异常,CompletableFuture通过whenComplete来减少阻塞和轮询（自动回调）if(e == null){//判断有没有异常System.out.println("计算完成,更新系统update value:"+v);}}).exceptionally(e ->{e.printStackTrace();System.out.println("异常情况:"+e.getCause()+"\t"+e.getMessage());return null;});System.out.println(Thread.currentThread().getName()+"线程先去忙其它任务");}catch (Exception e){e.printStackTrace();}finally {//关闭线程池threadPool.shutdown();}}
}

正常执行结果:

异常执行结果:

CompletableFuture优点总结:

• 异步任务结束时，会自动回调某个对象的方法；
• 主线程设置好回调后，不再关心异步任务的执行，异步任务之间可以顺序执行
• 异步任务出错时，会自动回调某个对象的方法

2.3.4 函数式接口串讲

函数式接口定义:

任何接口,如果只包含唯一一个抽象方法,那么它就是一个函数式接口.对于函数式接口,我们可以通过lambda表达式来创建该接口的对象.

更准确的来说接口定义时带有@FunctionalInterface注解的都是函数式接口

2.3.4.1 常见的函数式接口

• Biconsumer(Bi是英语词根代表两个的意思，我们要传入两个参数，在上面的案例中是v和e)

2.3.4.2 链式调用写法

@Accessors(chain = true)开启链式编程,需要lombok

public class Chain {public static void main(String[] args) {//-------------------老式写法------------
//        Student student = new Student();
//        student.setId(1);
//        student.setMajor("cs");
//        student.setName("小卡");new Student().setId(1).setName("大卡").setMajor("cs");} 
}@NoArgsConstructor
@AllArgsConstructor
@Data
@Accessors(chain = true)//开启链式编程
class Student{private int id;private String name;private String major;
}

2.3.4.3 join和get对比

功能几乎一样,区别在于编码时是否需要抛出异常

• get()方法会在编译期间会做异常的检查,因此需要抛出异常或者做异常处理
• join()方法不会在编译期间会做异常的检查

public class Chain {public static void main(String[] args) throws ExecutionException, InterruptedException {//抛出异常CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {return "hello 12345";});System.out.println(completableFuture.get());}}public class Chain {public static void main(String[] args)  {//抛出异常CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {return "hello 12345";});System.out.println(completableFuture.join());}
}

2.4 CompletableFuture电商比价案例

实战精讲-比价网站case:
1 需求说明
1.1 同一款产品，同时搜索出同款产品在各大电商平台的售价；
1.2 同一款产品，同时搜索出本产品在同一个电商平台下，各个入驻卖家售价是多少2 输出返回：
出来结果希望是同款产品的在不同地方的价格清单列表， 返回一个List
《mysql》in jd price is 88.05
《mysql》in dang dang price is 86.11
《mysql》in tao bao price is 90.433 解决方案，比对同一个商品在各个平台上的价格，要求获得一个清单列表
1   stepbystep   ， 按部就班， 查完京东查淘宝， 查完淘宝查天猫......
2   all in       ，万箭齐发，一口气多线程异步任务同时查询。。。

package site.zhourui.juc;import lombok.Getter;import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;class CompletableFutureNetMallDemo
{static List list = Arrays.asList(new NetMall("jd"),new NetMall("pdd"),new NetMall("taobao"),new NetMall("dangdangwang"),new NetMall("tmall"));//同步 ，step by step/*** List  ---->   List* @param list* @param productName* @return*/public static List getPriceByStep(List list,String productName){return list.stream().map(netMall -> String.format(productName + " in %s price is %.2f", netMall.getMallName(), netMall.calcPrice(productName))).collect(Collectors.toList());}//异步 ，多箭齐发/*** List  ---->List> --->   List* @param list* @param productName* @return*/public static List getPriceByASync(List list,String productName){return list.stream().map(netMall -> CompletableFuture.supplyAsync(() -> String.format(productName + " is %s price is %.2f", netMall.getMallName(), netMall.calcPrice(productName)))).collect(Collectors.toList()).stream().map(CompletableFuture::join).collect(Collectors.toList());}public static void main(String[] args){long startTime = System.currentTimeMillis();List list1 = getPriceByStep(list, "mysql");for (String element : list1) {System.out.println(element);}long endTime = System.currentTimeMillis();System.out.println("----costTime: "+(endTime - startTime) +" 毫秒");System.out.println();long startTime2 = System.currentTimeMillis();List list2 = getPriceByASync(list, "mysql");for (String element : list2) {System.out.println(element);}long endTime2 = System.currentTimeMillis();System.out.println("----costTime: "+(endTime2 - startTime2) +" 毫秒");}
}class NetMall
{@Getterprivate String mallName;public NetMall(String mallName){this.mallName = mallName;}public double calcPrice(String productName){//检索需要1秒钟try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }return ThreadLocalRandom.current().nextDouble() * 2 + productName.charAt(0);}
}

执行结果:

3. CompletableFuture常用方法

3.1 获得结果和触发计算

3.1.1 getNow

get之前介绍过了,这里介绍getNow

getNow相当于备用方案如果此时异步任务还没执行完成就使用getNow设置的默认值,如果完成了就是用任务返回值

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureAPIDemo
{public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException{ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 20, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(50), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());CompletableFuture future = CompletableFuture.supplyAsync(() -> {//暂停几秒钟线程//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }return 1;},threadPoolExecutor);//System.out.println(future.get());//System.out.println(future.get(2L,TimeUnit.SECONDS));//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }System.out.println(future.getNow(9999));threadPoolExecutor.shutdown();}
}

3.1.2 complete主动触发计算

当调用CompletableFuture.get()被阻塞的时候,complete方法就是结束阻塞并get()获取设置的complete里面的值.

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureAPIDemo
{public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException{ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 20, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(50), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());CompletableFuture future = CompletableFuture.supplyAsync(() -> {//暂停几秒钟线程//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }return 1;},threadPoolExecutor);//System.out.println(future.get());//System.out.println(future.get(2L,TimeUnit.SECONDS));//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
//        System.out.println(future.getNow(9999));System.out.println(future.complete(-44)+"\t"+future.get());threadPoolExecutor.shutdown();}
}

3.2 对计算结果进行处理

总结:

thenApply 和handle都是将线程船型化

但是thenApply 报异常后,之后的线程无法继续执行

handle 报异常后,之后的线程可以继续执行

thenApply 带一个参数即返回值,handle带两个参数多带一个异常

3.2.1 thenApply

出错了不会继续执行

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException{//当一个线程依赖另一个线程时用 thenApply 方法来把这两个线程串行化,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 -> {int age = 10/0; // 异常情况：那步出错就停在那步。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");// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }}
}

3.2.2 handle

出错了还是会继续执行

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException{//当一个线程依赖另一个线程时用 handle 方法来把这两个线程串行化,// 异常情况：有异常也可以往下一步走，根据带的异常参数可以进一步处理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");// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }}
}

3.3 对计算结果进行消费

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

3.3.1 thenAccept

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException{CompletableFuture.supplyAsync(() -> {return 1;}).thenApply(f -> {return f + 2;}).thenApply(f -> {return f + 3;}).thenApply(f -> {return f + 4;}).thenAccept(r -> System.out.println(r));}
}

3.3.2 Code之任务之间的顺序执行

thenRun相当于不需要结果也不会返回值

3.4 对计算速度进行选用

谁快用谁

3.4.1 applyToEither

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException{CompletableFuture completableFuture1 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");//暂停几秒钟线程try { TimeUnit.SECONDS.sleep(2); } catch (InterruptedException e) { e.printStackTrace(); }return 10;});CompletableFuture completableFuture2 = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }return 20;});CompletableFuture thenCombineResult = completableFuture1.applyToEither(completableFuture2,f -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in ");return f + 1;});System.out.println(Thread.currentThread().getName() + "\t" + thenCombineResult.get());}
}

3.5 对计算结果进行合并

• 两个CompletionStage任务都完成后，最终能把两个任务的结果一起交给thenCombine 来处理
• 先完成的先等着，等待其它分支任务

3.5.1 thenCombine

package site.zhourui.juc.cf;import java.util.concurrent.*;public class CompletableFutureBuildDemo2 {public static void main(String[] args) throws ExecutionException, InterruptedException{CompletableFuture thenCombineResult = CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in 1");return 10;}).thenCombine(CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in 2");return 20;}), (x,y) -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in 3");return x + y;}).thenCombine(CompletableFuture.supplyAsync(() -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in 4");return 30;}),(a,b) -> {System.out.println(Thread.currentThread().getName() + "\t" + "---come in 5");return a + b;});System.out.println("-----主线程结束，END");System.out.println(thenCombineResult.get());// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:try { TimeUnit.SECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); }}
}