java 8 新特性功能和用法介绍03---CompletableFuture基本用法介绍

Java8 CompletableFuture

java 8 新增CompletableFuture类简化异步编程的复杂性，提供函数式编程的能力，并且可以通过回调的方式处理计算结果。同时解决了
传统异步编程Future模式的缺点，对于异步执行结果，只能通过等待get操作以及轮询isDone去判断Future是否完成，是非常耗费CPU资源。而CompletableFuture
弥补了Future模式的缺点，在异步的任务完成后，需要用其结果继续操作时，无需等待。可以直接通过thenAccept、thenApply、thenCompose等方式
将前面异步处理的结果交给另外一个异步事件处理线程来处理。
本文将结束如下内容：

• CompletableFuture 常用API介绍
• CompletableFuture 与并行流区别

CompletableFuture 常用API介绍

主要分CompletableFuture创建、转换、消费、组合、错误处理等API使用。

CompletableFuture创建

public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
public static <U> CompletableFuture<U> completedFuture(U value)

上面方法，没有带Executor参数，默认使用 ForkJoinPool.commonPool()线程池执行异步代码。如果带表示可以自定义线程池大小

代码示例：

 public void testRunAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> run = CompletableFuture.runAsync(() -> System.out.println("Hello World!"));
        run.get();
    }

    public void testSupplyAsync()throws ExecutionException, InterruptedException {
        ExecutorService service = Executors.newFixedThreadPool(10);
        CompletableFuture<Cat> cat = CompletableFuture.supplyAsync(() -> new Cat("tom", "black"), service);
        System.out.println(cat.get().getName());
    }
    
       public CompletableFuture<List<Cat>> testCompletedFuture() {
            List<Cat> cats = Arrays.asList(new Cat("tom", "black"), new Cat("jack", "white")
                    , new Cat("lily", "black"));
            return CompletableFuture.completedFuture(cats);
        }

上面代码中runAsync是没有返回值，而supplyAsync以及completedFuture静态辅助方法是有返回值，通常使用他们构造CompletableFuture对象

CompletableFuture转换

public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn,Executor executor);

上面方法的输入上一个阶段的计算后输出的结果，其中带Async表示使用默认线程池进行异步处理，不带Async表示使用原来线程进行同步处理，后面章节同理。

代码示例：

    private CompletableFuture<List<Cat>> testCompletedFuture() {
           List<Cat> cats = Arrays.asList(new Cat("tom", "black"), new Cat("jack", "white")
                   , new Cat("lily", "black"));
           return CompletableFuture.completedFuture(cats);
       }
   
       public void testThenApplyAsync(){
           //统计黑猫个数
           CompletableFuture<Long> countStage = testCompletedFuture().thenApplyAsync(cats -> cats.stream().filter(cat -> "black".equals(cat.getColour())).count());
           System.out.println(countStage.join());
       }

CompletableFuture 消费

public CompletionStage<Void> thenAccept(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action);
public CompletionStage<Void> thenAcceptAsync(Consumer<? super T> action,Executor executor);

方法入参consumer，只消费结果，不产生新的计算值

 public void testThenAcceptAsync() {
        testCompletedFuture().thenAcceptAsync(cats -> {
            long count = cats.stream().map(Cat::getName).distinct().count();
            System.out.println("猫名的个数：" + count);
        });
    }

CompletableFuture 组合

public <U,V> CompletionStage<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn);
public <U,V> CompletionStage<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn,Executor executor);

该方法两个CompletionStage是并行执行的，它们之间并没有先后依赖顺序，other并不会等待先前的CompletableFuture执行完毕后再执行。

 public List<People> getPeopleList() {
        try {
            System.out.println(Thread.currentThread().getName() + ">>>处理查询人列表逻辑");
            Thread.sleep(1000L);
        } catch (Exception e) {
            throw new RuntimeException("系统运行异常");
        }
        return Arrays.asList(new People("张三", 17, Arrays.asList(new Cat("tom", "black")
                , new Cat("jack", "white"))), new People("李四", 19, Arrays.asList(new Cat("a", "black")
                , new Cat("b", "white"))));
    }

 public List<Cat> getCatList() {
        try {
            System.out.println(Thread.currentThread().getName() + ">>>处理查询猫列表逻辑");
            Thread.sleep(2000L);
        } catch (Exception e) {
              throw new RuntimeException("系统运行异常");
        }
        return Arrays.asList(new Cat("tom", "black"), new Cat("jack", "white")
                , new Cat("lily", "black"));
    }
    
public void testThenCombineAsync() {
        List<Cat> cats = Arrays.asList(new Cat("小智", "black"), new Cat("阿黄", "white")
                , new Cat("小黑", "black"));
        CompletableFuture<List<Cat>> catStage1 = CompletableFuture.completedFuture(cats);
        CompletableFuture<List<Cat>> catsStage2 = testCompletedFuture();

        Long count = catStage1.thenCombineAsync(catsStage2, (first, second) -> {
            List<Cat> result = new ArrayList<>();
            result.addAll(first);
            result.addAll(second);
            return result.stream().filter(cat -> "black".equals(cat.getColour())).count();
        }).join();
        System.out.println(count);

另外thenAcceptBoth、runAfterBoth、acceptEither大体上类似，总结如下，：
A,B -> C
thenCombineAsync : 结合上一步（A）结果和本次（B）的结果，然后进行转换
thenAcceptBothAsync : 结合上一步结果和本次的结果，然后进行消耗。消耗计算是在本次线程（本次线程就是上文提到的原来的线程）上执行；而Async是在线程池上执行。下面以此类推
runAfterBothAsync : 不关心上一步和本次的结果，只关心它们运算完毕，然后进行下步操作
acceptEitherAsync : 上一步和本次哪个快就用哪个的结果，然后消费

CompletableFuture 扁平化

public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn);
public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor);

方法接受一个Function作为参数，这个Function的输入是当前的CompletableFuture的计算值，返回结果将是一个新的CompletableFuture，这个新的CompletableFuture会组合原来的CompletableFuture和函数返回的CompletableFuture。
通俗说将其旧扁平化新的CompletableFuture

 public void testThenComposeAsync() {
        CompletableFuture<List<People>> peopleList = CompletableFuture.completedFuture(getPeopleList());
        peopleList.thenComposeAsync(people -> {
            //假设查询该人养了几只猫，并且是异步查询
            return CompletableFuture.completedFuture(getCatList());
        }).thenAccept(cats -> System.out.println("养了" + cats.size() + "只")).join();
    }

CompletableFuture 计算完成后处理

public CompletableFuture<T> 	whenComplete(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> 	whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
public CompletableFuture<T> 	whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)
public CompletableFuture<T>     exceptionally(Function<Throwable,? extends T> fn)
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

whenCompleteAsync方法接受一个BiConsumer消费方法，当计算完成，或者抛出异常的时候，可以对其结果或异常进行消费,不影响原来计算结果
exceptionally接受fn,当运行时出现了异常，可以通过exceptionally进行补偿（消化）
handleAsync接受BiFn,当计算完成，或者抛出异常的时候，可以对其结果或异常进行转换，如果发生异常时可以中断，改变原来计算结果

public void testWhenCompleteAsync() {
        CompletableFuture<List<People>> peopleList = CompletableFuture.completedFuture(getPeopleList());
        //使用complete仅对最终计算结果进行消费，不影响计算结果
        peopleList.whenCompleteAsync((people, t) -> {
            if (t != null) {
                System.out.println(t.getMessage());
            } else {
                people.stream().collect(Collectors.toMap(People::getName, People::getCats)).forEach((name, cats) -> {
                    System.out.println(name + "养了" + cats.size() + "只猫");
                });
            }
        }).join();

        //exceptionally对异常情况进行补偿
        peopleList.exceptionally(e -> {
            //当出现异常时，记录异常并返回空对象
            System.out.println(e.getMessage());
            return Collections.emptyList();
        }).join();
        
        //handleAsync对异常进行终止，改变原来计算结果
        List<People> peoples = peopleList.handleAsync((p, t) -> {
            if (t != null) {
                return Collections.<People>emptyList();
            }
            return p;
        }).join();
    }

CompletableFuture 辅助方法AnyOf,AllOf

public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) 

allOf方法所有给定的CompletableFutures都完成时,返回的新CompletableFuture。
anyOf方法任何给定的CompletableFutures都完成时,返回的新CompletableFuture。

   public void testAllOf(){
        CompletableFuture<List<Cat>> catStage = CompletableFuture.completedFuture(getCatList());
        CompletableFuture<List<People>> peopleStage = CompletableFuture.completedFuture(getPeopleList());
        CompletableFuture.allOf(catStage,peopleStage).thenAcceptAsync(all->{
            List<Cat> cats = catStage.join();
            System.out.println(cats.size());
            List<People> peoples = peopleStage.join();
            System.out.println(peoples.size());
        }).join();
    }

   public void testAnyOf(){
        CompletableFuture<List<Cat>> catStage = CompletableFuture.completedFuture(getCatList());
        CompletableFuture<List<People>> peopleStage = CompletableFuture.completedFuture(getPeopleList());
        Object result = CompletableFuture.anyOf(catStage, peopleStage).join();
    }

CompletableFuture 与 parallelStream区别

parallelStream/O的任务，比较简单就能实现多线程处理，而且充分利用了计算机的CPU资源。
因为parallelStream底层使用的是默认的ForkJoinPool，该线程池的线程数和CPU的核心数目一致，如果线程中需要长时间的I/O，就使得其他需要使用并行流的任务阻塞。
parallelStream如果计算量比较大的话，可以采用自定义ForkJoinPool的方式，来增大线程池的线程数。

CompletableFuture适合在并行的工作单元涉及等待I/O的操作，如比较耗时的网络请求调用。而且CompletableFuture比较灵活，有多个静态方法来完成异步结果返回之后的操作。
值得注意的是，CompletableFuture的join方法相当于Future的get方法，都会阻塞住调用线程。

本文地址：https://blog.csdn.net/weixin_43946605/article/details/110224973