39、CompletableFuture的使用

API使用

• 简单执行

public static void test1() throws Exception {CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {return "hello";});System.out.println(Thread.currentThread().getName() + ":" +stringCompletableFuture.get());}

• 结果作为下一个的参数

/*** 结果作为下一个参数  无返回值* @throws Exception*/public static void test3() throws Exception {CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {return "hello";});CompletableFuture<Void> stringCompletableFuture1 = stringCompletableFuture.thenAccept(s -> {System.out.println(s + " world");});System.out.println(Thread.currentThread().getName() + ":" +stringCompletableFuture1.get());}/*** 结果作为下一个参数  又返回值* @throws ExecutionException* @throws InterruptedException*/public static void test2() throws Exception {CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {return "hello";});CompletableFuture<String> stringCompletableFuture1 = stringCompletableFuture.thenApply(s -> {return s + " world";});System.out.println(Thread.currentThread().getName() + ":" +stringCompletableFuture1.get());}

• thenCompose的使用（类似于flatmap）

public static void test4() throws Exception {ProductService productService = new ProductService();ProductDetailService productDetailService = new ProductDetailService();
//      同一个线程执行
//        CompletableFuture<Product> productCompletableFuture = CompletableFuture.supplyAsync(() -> {
//            String name = productService.getProduct(1);
//            Product product = new Product();
//            product.setName(name);
//            return product;
//        }).thenApply(product -> {
//            String desc = productDetailService.getProduct(1);
//            product.setDesc(desc);
//            return product;
//        });
//        System.out.println(Thread.currentThread().getName() + ":" +productCompletableFuture.get());//不同的线程执行CompletableFuture<CompletableFuture<Product>> completableFutureCompletableFuture = CompletableFuture.supplyAsync(() -> {String name = productService.getProduct(1);Product product = new Product();product.setName(name);return product;}).thenApply(new Function<Product, CompletableFuture<Product>>() {@Overridepublic CompletableFuture<Product> apply(Product product) {return CompletableFuture.supplyAsync(() -> {String desc = productDetailService.getProduct(1);product.setDesc(desc);return product;});}});System.out.println(Thread.currentThread().getName() + ":" +completableFutureCompletableFuture.get().get());
////不同的线程执行
//        CompletableFuture<Product> productCompletableFuture1 = CompletableFuture.supplyAsync(() -> {
//            String name = productService.getProduct(1);
//            Product product = new Product();
//            product.setName(name);
//            return product;
//        }).thenCompose((Function<Product, CompletableFuture<Product>>) product -> CompletableFuture.supplyAsync(() -> {
//            String desc = productDetailService.getProduct(1);
//            product.setDesc(desc);
//            return product;
//        }));
//        System.out.println(Thread.currentThread().getName() + ":" +productCompletableFuture1.get());}

• 多个Future 如何处理他们的一些聚合关系呢
  • allOf
    • 返回值是CompletableFuture<Void>类型
    • 因为allOf没有返回值，所以通过thenApply获取CompletableFuture的执行结果
    • 使用场景：请求多个接口 聚合结果
  • anyOf
    • 只要有任意一个CompletableFuture结束，就可以做接下来的事情，不像allOf要等待所有的CompletableFuture都结束才做接下来的事情
    • 每个CompletableFuture的返回值类型都不相同，无法判断是什么类型
    • 所以anyOf返回值是CompletableFuture<Object>类型
    • 使用场景：多个机房 机器配置以及压力不一样 请求同一个接口 返回参数也一样 ，使用anyOf同步请求所有机房，然后谁最新返回 就直接使用返回结果

 /*** 多个Future 如何处理他们的一些聚合关系呢* allOf  anyOf* @throws Exception*/public static void test6() throws Exception {CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + ":" + "future1");return "future1";});CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(2000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + ":" + "future2");return "future2";});CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(3000);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + ":" + "future3");return "future3";});//等待所有任务完成
//        CompletableFuture<Void> voidCompletableFuture = CompletableFuture.allOf(future1, future2, future3);
//        System.out.println("begin" + LocalDateTime.now());
//        voidCompletableFuture.join();
//        if (voidCompletableFuture.isDone()){
//            System.out.println("任务全部完成");
//        }
//        System.out.println("end" + LocalDateTime.now());System.out.println("-----------------------------");//等待任何一个任务完成CompletableFuture<Object> objectCompletableFuture = CompletableFuture.anyOf(future1, future2, future3);System.out.println("begin" + LocalDateTime.now());objectCompletableFuture.join();if (objectCompletableFuture.isDone()){System.out.println("有任务完成");}System.out.println("end" + LocalDateTime.now());}

案例实战

• 背景
  • 微服务架构，接口单一职责，一个页面打开会涉及多个模块需要同时调用
  • 由于需要同时建立多个连接，中间会有性能损耗，部分页面需要使用聚合接口
  • 则可以使用CompletableFuture聚合多个响应结果一次性返回
    • 优点
      • 减少建立连接数量
      • 网关和服务端可以处理更多的连接
    • 缺点
      • 如果接口性能差异大，则容易性能好的接口被性能查的拖垮
      • 需要开发更大接口，数据量大则需要更大的带宽
  • 其他场景
    • 爬虫业务多个URL并行爬取
    • 商品详情页信息组装
    • 业务聚合接口：用户信息（积分基本信息，权限）