1、CompletableFuture介绍
CompletableFuture对象是JDK1.8版本新引入的类，这个类实现了两个接口，一个是Future接口，一个是CompletionStage接口。
CompletionStage接口是JDK1.8版本提供的接口，用于异步执行中的阶段处理，CompletionStage定义了一组接口用于在一个阶段执行结束之后，要么继续执行下一个阶段，要么对结果进行转换产生新的结果等，一般来说要执行下一个阶段都需要上一个阶段正常完成，这个类也提供了对异常结果的处理接口
2、CompletableFuture的API
2.1 提交任务
在CompletableFuture中提交任务有以下几种方式：
1public static CompletableFuture<Void> runAsync(Runnable runnable) 2public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) 3public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 4public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) 5
这四个方法都是用来提交任务的，不同的是supplyAsync提交的任务有返回值，runAsync提交的任务没有返回值。两个接口都有一个重载的方法，第二个入参为指定的线程池，如果不指定，则默认使用ForkJoinPool.commonPool()线程池。在使用的过程中尽量根据不同的业务来指定不同的线程池，方便对不同线程池进行监控，同时避免业务共用线程池相互影响。
2.2 结果转换
2.2.1 thenApply
1public <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) 2public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) 3public <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) 4
thenApply这一组函数入参是Function，意思是将上一个CompletableFuture执行结果作为入参，再次进行转换或者计算，重新返回一个新的值。
2.2.2 handle
1public <U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn) 2public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn) 3public <U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor) 4
handle这一组函数入参是BiFunction，该函数式接口有两个入参一个返回值，意思是处理上一个CompletableFuture的处理结果，同时如果有异常，需要手动处理异常。
2.2.3 thenRun
1public CompletableFuture<Void> thenRun(Runnable action) 2public CompletableFuture<Void> thenRunAsync(Runnable action) 3public CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor) 4
thenRun这一组函数入参是Runnable函数式接口，该接口无需入参和出参，这一组函数是在上一个CompletableFuture任务执行完成后，在执行另外一个接口，不需要上一个任务的结果，也不需要返回值，只需要在上一个任务执行完成后执行即可。
2.2.4 thenAccept
1public CompletableFuture<Void> thenAccept(Consumer<? super T> action) 2public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) 3public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor) 4
thenAccept这一组函数的入参是Consumer，该函数式接口有一个入参，没有返回值，所以这一组接口的意思是处理上一个CompletableFuture的处理结果，但是不返回结果。
2.2.5 thenAcceptBoth
1public <U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) 2public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action) 3public <U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor) 4
thenAcceptBoth这一组函数入参包括CompletionStage以及BiConsumer，CompletionStage是JDK1.8新增的接口，在JDK中只有一个实现类：CompletableFuture，所以第一个入参就是CompletableFuture，这一组函数是用来接受两个CompletableFuture的返回值，并将其组合到一起。BiConsumer这个函数式接口有两个入参，并且没有返回值，BiConsumer的第一个入参就是调用方CompletableFuture的执行结果，第二个入参就是thenAcceptBoth接口入参的CompletableFuture的执行结果。所以这一组函数意思是将两个CompletableFuture执行结果合并到一起。
2.2.6 thenCombine
1public <U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 2public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn) 3public <U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor) 4
thenCombine这一组函数和thenAcceptBoth类似，入参都包含一个CompletionStage，也就是CompletableFuture对象，意思也是组合两个CompletableFuture的执行结果，不同的是thenCombine的第二个入参为BiFunction，该函数式接口有两个入参，同时有一个返回值。所以与thenAcceptBoth不同的是，thenCombine将两个任务结果合并后会返回一个全新的值作为出参。
2.2.7 thenCompose
1public <U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn) 2public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn) 3public <U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor) 4
thenCompose这一组函数意思是将调用方的执行结果作为Function函数的入参，同时返回一个新的CompletableFuture对象。
2.3 回调方法
1public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) 2public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action) 3public CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor) 4
whenComplete方法意思是当上一个CompletableFuture对象任务执行完成后执行该方法。BiConsumer函数式接口有两个入参没有返回值，这两个入参第一个是CompletableFuture任务的执行结果，第二个是异常信息。表示处理上一个任务的结果，如果有异常，则需要手动处理异常，与handle方法的区别在于，handle方法的BiFunction是有返回值的，而BiConsumer是没有返回值的。
以上方法都有一个带有Async的方法，带有Async的方法表示是异步执行的，会将该任务放到线程池中执行，同时该方法会有一个重载的方法，最后一个参数为Executor，表示异步执行可以指定线程池执行。为了方便进行控制，最好在使用CompletableFuture时手动指定我们的线程池。
2.4 异常处理
1public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn) 2
exceptionally是用来处理异常的，当任务抛出异常后，可以通过exceptionally来进行处理，也可以选择使用handle来进行处理，不过两者有些不同，hand是用来处理上一个任务的结果，如果有异常情况，就处理异常。而exceptionally可以放在CompletableFuture处理的最后，作为兜底逻辑来处理未知异常。
2.5 获取结果
1public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) 2public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) 3
allOf是需要入参中所有的CompletableFuture任务执行完成，才会进行下一步；
anyOf是入参中任何一个CompletableFuture任务执行完成都可以执行下一步。
1public T get() throws InterruptedException, ExecutionException 2public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 3public T getNow(T valueIfAbsent) 4public T join() 5
get方法一个是不带超时时间的，一个是带有超时时间的。
getNow方法则是立即返回结果，如果还没有结果，则返回默认值，也就是该方法的入参。
join方法是不带超时时间的等待任务完成。
3、CompletableFuture原理
join方法同样表示获取结果，但是join与get方法有什么区别呢。
1public T join() { 2 Object r; 3 return reportJoin((r = result) == null ? waitingGet(false) : r); 4} 5 6public T get() throws InterruptedException, ExecutionException { 7 Object r; 8 return reportGet((r = result) == null ? waitingGet(true) : r); 9} 10 11public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { 12 Object r; 13 long nanos = unit.toNanos(timeout); 14 return reportGet((r = result) == null ? timedGet(nanos) : r); 15} 16 17public T getNow(T valueIfAbsent) { 18 Object r; 19 return ((r = result) == null) ? valueIfAbsent : reportJoin(r); 20} 21
以上是CompletableFuture类中两个方法的代码，可以看到两个方法几乎一样。区别在于reportJoin/reportGet，waitingGet方法是一致的，只不过参数不一样，我们在看下reportGet与reportJoin方法。
1private static <T> T reportGet(Object r) 2 throws InterruptedException, ExecutionException { 3 if (r == null) // by convention below, null means interrupted 4 throw new InterruptedException(); 5 if (r instanceof AltResult) { 6 Throwable x, cause; 7 if ((x = ((AltResult)r).ex) == null) 8 return null; 9 if (x instanceof CancellationException) 10 throw (CancellationException)x; 11 if ((x instanceof CompletionException) && 12 (cause = x.getCause()) != null) 13 x = cause; 14 throw new ExecutionException(x); 15 } 16 @SuppressWarnings("unchecked") T t = (T) r; 17 return t; 18 } 19
1private static <T> T reportJoin(Object r) { 2 if (r instanceof AltResult) { 3 Throwable x; 4 if ((x = ((AltResult)r).ex) == null) 5 return null; 6 if (x instanceof CancellationException) 7 throw (CancellationException)x; 8 if (x instanceof CompletionException) 9 throw (CompletionException)x; 10 throw new CompletionException(x); 11 } 12 @SuppressWarnings("unchecked") T t = (T) r; 13 return t; 14 } 15
可以看到这两个方法很相近，reportGet方法判断了r对象是否为空，并抛出了中断异常，而reportJoin方法没有判断，同时reportJoin抛出的都是运行时异常，所以join方法也是无需手动捕获异常的。
我们在看下waitingGet方法
1private Object waitingGet(boolean interruptible) { 2 Signaller q = null; 3 boolean queued = false; 4 int spins = -1; 5 Object r; 6 while ((r = result) == null) { 7 if (spins < 0) 8 spins = SPINS; 9 else if (spins > 0) { 10 if (ThreadLocalRandom.nextSecondarySeed() >= 0) 11 --spins; 12 } 13 else if (q == null) 14 q = new Signaller(interruptible, 0L, 0L); 15 else if (!queued) 16 queued = tryPushStack(q); 17 else if (interruptible && q.interruptControl < 0) { 18 q.thread = null; 19 cleanStack(); 20 return null; 21 } 22 else if (q.thread != null && result == null) { 23 try { 24 ForkJoinPool.managedBlock(q); 25 } catch (InterruptedException ie) { 26 q.interruptControl = -1; 27 } 28 } 29 } 30 if (q != null) { 31 q.thread = null; 32 if (q.interruptControl < 0) { 33 if (interruptible) 34 r = null; // report interruption 35 else 36 Thread.currentThread().interrupt(); 37 } 38 } 39 postComplete(); 40 return r; 41 } 42
该waitingGet方法是通过while的方式循环判断是否任务已经完成并产生结果，如果结果为空，则会一直在这里循环，这里需要注意的是在这里初始化了一下spins=-1，当第一次进入while循环的时候，spins是-1，这时会将spins赋值为一个常量，该常量为SPINS。
1private static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ? 2 1 << 8 : 0); 3
这里判断可用CPU数是否大于1，如果大于1，则该常量为 1<< 8，也就是256，否则该常量为0。
第二次进入while循环的时候，spins是256大于0，这里做了减一的操作，下次进入while循环，如果还没有结果，依然是大于0继续做减一的操作，此处用来做短时间的自旋等待结果，只有当spins等于0，后续会进入正常流程判断。
我们在看下timedGet方法的源码
1private Object timedGet(long nanos) throws TimeoutException { 2 if (Thread.interrupted()) 3 return null; 4 if (nanos <= 0L) 5 throw new TimeoutException(); 6 long d = System.nanoTime() + nanos; 7 Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0 8 boolean queued = false; 9 Object r; 10 // We intentionally don't spin here (as waitingGet does) because 11 // the call to nanoTime() above acts much like a spin. 12 while ((r = result) == null) { 13 if (!queued) 14 queued = tryPushStack(q); 15 else if (q.interruptControl < 0 || q.nanos <= 0L) { 16 q.thread = null; 17 cleanStack(); 18 if (q.interruptControl < 0) 19 return null; 20 throw new TimeoutException(); 21 } 22 else if (q.thread != null && result == null) { 23 try { 24 ForkJoinPool.managedBlock(q); 25 } catch (InterruptedException ie) { 26 q.interruptControl = -1; 27 } 28 } 29 } 30 if (q.interruptControl < 0) 31 r = null; 32 q.thread = null; 33 postComplete(); 34 return r; 35 } 36
timedGet方法依然是通过while循环的方式来判断是否已经完成，timedGet方法入参为一个纳秒值，并通过该值计算出一个deadline截止时间，当while循环还未获取到任务结果且已经达到截止时间，则抛出一个TimeoutException异常。
4、CompletableFuture实现多线程任务
这里我们通过CompletableFuture来实现一个多线程处理异步任务的例子。
这里我们创建10个任务提交到我们指定的线程池中执行，并等待这10个任务全部执行完毕。
每个任务的执行流程为第一次先执行加法，第二次执行乘法，如果发生异常则返回默认值，当10个任务执行完成后依次打印每个任务的结果。
1public void demo() throws InterruptedException, ExecutionException, TimeoutException { 2 // 1、自定义线程池 3 ExecutorService executorService = new ThreadPoolExecutor(5, 10, 4 60L, TimeUnit.SECONDS, 5 new LinkedBlockingQueue<>(100)); 6 7 // 2、集合保存future对象 8 List<CompletableFuture<Integer>> futures = new ArrayList<>(10); 9 for (int i = 0; i < 10; i++) { 10 int finalI = i; 11 CompletableFuture<Integer> future = CompletableFuture 12 // 提交任务到指定线程池 13 .supplyAsync(() -> this.addValue(finalI), executorService) 14 // 第一个任务执行结果在此处进行处理 15 .thenApplyAsync(k -> this.plusValue(finalI, k), executorService) 16 // 任务执行异常时处理异常并返回默认值 17 .exceptionally(e -> this.defaultValue(finalI, e)); 18 // future对象添加到集合中 19 futures.add(future); 20 } 21 22 // 3、等待所有任务执行完成，此处最好加超时时间 23 CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get(5, TimeUnit.MINUTES); 24 for (CompletableFuture<Integer> future : futures) { 25 Integer num = future.get(); 26 System.out.println("任务执行结果为：" + num); 27 } 28 System.out.println("任务全部执行完成！"); 29 } 30 31 private Integer addValue(Integer index) { 32 System.out.println("第" + index + "个任务第一次执行"); 33 if (index == 3) { 34 int value = index / 0; 35 } 36 return index + 3; 37 } 38 39 private Integer plusValue(Integer index, Integer num) { 40 System.out.println("第" + index + "个任务第二次执行，上次执行结果：" + num); 41 return num * 10; 42 } 43 44 private Integer defaultValue(Integer index, Throwable e) { 45 System.out.println("第" + index + "个任务执行异常！" + e.getMessage()); 46 e.printStackTrace(); 47 return 10; 48 } 49
作者：京东物流 丁冬
来源：京东云开发者社区 自猿其说Tech