1.创建异步对象

CompletableFuture提供四个静态方法来创建一个异步操作。

//runAsync，传入Runnable ，返回CompletableFuture<Void> ，无返回值，Executor 可以指定一个线程池
public static CompletableFuture<Void> runAsync(Runnable runnable) 
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor) 
//supplyAsync，有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor) 

runAsync举例：

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        System.out.println("main start");
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
        }, executor);
        System.out.println("main end");
    }
}

main start
main end
当前线程：pool-1-thread-1
运行结果：5

supplyAsync举例：

pblic class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 2;
            System.out.println("运行结果：" + i);
            return i;
        }, executor);
        Integer result = future.get();
        System.out.println("main end,返回结果：" + result);
    }
}

main start
当前线程：pool-1-thread-1
运行结果：5
main end,返回结果：5

2.计算完成时回调方法

//上一个任务完成以后，再用同一个线程去调用下一个任务
public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
//上一个任务完成以后，再用异步的方式去调用下一个任务
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor);
//处理异常情况
public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) 

whenComplete例子（无异常）：

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 /0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).whenComplete((res, exception) -> {
            System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
        });
        System.out.println("main end");
    }
}

main start
main end
当前线程：pool-1-thread-1
运行结果：2
异步任务成功完成了。。结果是：2,异常是：null

whenComplete例子（有异常）：

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).whenComplete((res, exception) -> {
            System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
        });
        System.out.println("main end");
    }
}

main start
main end
当前线程：pool-1-thread-1
异步任务成功完成了。。结果是：null,异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero

exceptionally例子：

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).whenComplete((res, exception) -> {
            // 虽然可以得到异常信息，但是无法修改返回数据
            System.out.println("异步任务成功完成了。。结果是：" + res + ",异常是：" + exception);
        }).exceptionally(throwable -> {
            // 可以处理异常，同时在异常出现后返回一个值
            return 10;
        });
        System.out.println("main end,返回结果："+future.get());
    }
}

main start
当前线程：pool-1-thread-1
异步任务成功完成了。。结果是：null,异常是：java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
main end,返回结果：10

3.handle方法

和complete一样，可以对结果做最后的处理（可以处理异常），可以改变返回值

public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) 

例子

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");
       CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 0;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).handle((res, exception) -> {
            if (res != null) {
                return res * 2;
            }
            if (exception != null) {
                return 0;
            }
            return 0;
        });
        System.out.println("main end,返回结果：" + future.get());
    }
}

main start
当前线程：pool-1-thread-1
main end,返回结果：0

4.线程串行化方法

//thenApply：当一个线程依赖另一个线程时，获取上一个任务返回的结果 ，并返回当前任务的返回值。
//带有Async默认是异步执行任务。可指定线程池
public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) 
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

//thenAccept：消费处理结果 。接收任务的处理结果 ，并消费处理，无返回结果 
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) 
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) 
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)

//thenRun：只要上面的任务执行完成，就开始执行thenRun的后续操作。
public CompletableFuture<Void> thenRun(Runnable action)
public CompletableFuture<Void> thenRunAsync(Runnable action) 
public CompletableFuture<Void> thenRunAsync(Runnable action,Executor executor) 

举例

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        System.out.println("main start");
        
        /*
         *  thenRunAsync无法获取上一步执行结果
         *  如果上一任务抛出异常，thenRunAsync也不会执行
         */
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).thenRunAsync(() -> {
            System.out.println("任务2启动了");
        }, executor);
        System.out.println("main end");
    }
}

main start
main end
当前线程：pool-1-thread-1

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args)  {
        System.out.println("main start");

        /*
         * thenAccept可以获取上一步返回结果
         */
        CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).thenAccept(res -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            System.out.println("任务2启动了，上一任务执行结果：" + res);
        });
        System.out.println("main end");
    }
}

main start
main end
当前线程：pool-1-thread-1
运行结果：2
当前线程：pool-1-thread-1
任务2启动了，上一任务执行结果：2

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");

        /*
         * thenApplyAsync 带有返回值，可以返回结果 
         */
        CompletableFuture<Object> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("当前线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("运行结果：" + i);
            return i;
        }, executor).thenApplyAsync(res -> {
            System.out.println("任务2启动了,上一步结果：" + res);
            return "hello " + res;
        }, executor);
        System.out.println("main end,最终返回结果：" + future.get());
    }
}

main start
当前线程：pool-1-thread-1
运行结果：2
任务2启动了,上一步结果：2
main end,最终返回结果：hello 2

5.两任务组合——任务都要完成

// thenCombine：组合两个future,获取两个future的返回结果，并返回当前任务的返回值 
public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) 
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn) 
public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other,BiFunction<? super T,? super U,? extends V> fn, Executor executor)

// thenAcceptBoth：组合两个future，获取两个future任务的返回结果 ，然后处理任务，没有返回值。
public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action) 
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action)
public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other,BiConsumer<? super T, ? super U> action, Executor executor) 

// runAfterBoth：组合两个future，不需要获取future的结果 ，只需要两个future处理完任务后，处理该任务。
public CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action) 
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action) 
public CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other,Runnable action,Executor executor) 

例子

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");

        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("任务1结束，结果：" + i);
            return i;
        }, executor);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程：" + Thread.currentThread().getName());
            System.out.println("任务2结束");
            return "hello";
        }, executor);
        //任务3会等任务1和任务2结束以后再执行。
        future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
        System.out.println("main end");
    }
}

main start
main end
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
任务3开始。

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");

        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("任务1结束，结果：" + i);
            return i;
        }, executor);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程：" + Thread.currentThread().getName());
            System.out.println("任务2结束");
            return "hello";
        }, executor);
//        future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
        future01.thenAcceptBothAsync(future02, (f1, f2) -> System.out.println("任务3开始。任务1结果：" + f1 + "   任务2结果：" + f2), executor);
        System.out.println("main end");
    }
}

main start
main end
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
任务3开始。任务1结果：2 任务2结果：hello

public class completable {
    //创建一个线程池
    public static ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println("main start");

        CompletableFuture<Integer> future01 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1线程：" + Thread.currentThread().getName());
            int i = 10 / 4;
            System.out.println("任务1结束，结果：" + i);
            return i;
        }, executor);
        CompletableFuture<String> future02 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2线程：" + Thread.currentThread().getName());
            System.out.println("任务2结束");
            return "hello";
        }, executor);
//        future01.runAfterBothAsync(future02,()-> System.out.println("任务3开始。"),executor);
//        future01.thenAcceptBothAsync(future02, (f1, f2) -> System.out.println("任务3开始。任务1结果：" + f1 + "   任务2结果：" + f2), executor);
        CompletableFuture<String> future = future01.thenCombineAsync(future02, (f1, f2) -> f1 + ":" + f2 + "->haha", executor);
        System.out.println("main end,最终返回结果:"+future.get());
    }
}

main start
任务1线程：pool-1-thread-1
任务1结束，结果：2
任务2线程：pool-1-thread-2
任务2结束
main end,最终返回结果:2:hello->haha

6.两任务组合——任意一个任务完成

// applyToEither: 两个任务有一个执行完成，获取它的返回值。处理任务并有新的返回值。
public <U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn) 
public <U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn,Executor executor) 

// acceptEither：两个任务有一个执行完成,获取它的返回值，处理任务，没有新的返回值。
public CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action) 
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
public CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action,Executor executor)

// runAfterEither：两个任务有一个执行完成，不需要获取返回值，处理第三个任务，也没有返回值。
public CompletableFuture<Void> runAfterEither(CompletionStage<?> other,Runnable action) 
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action)
public CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other,Runnable action,Executor etxecutor) 

7.多任务组合

// allOf：等待所有任务完成
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)

// anyOf：只要有一个任务完成
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)