CompletableFuture真香，可以替代CountDownLatch！

[[420499]]

创新互联建站主营阿图什网站建设的网络公司,主营网站建设方案,成都app开发,阿图什h5小程序开发搭建,阿图什网站营销推广欢迎阿图什等地区企业咨询

在对类的命名这篇长文中，我们提到了Future和Promise。

Future相当于一个占位符，代表一个操作将来的结果。一般通过get可以直接阻塞得到结果，或者让它异步执行然后通过callback回调结果。

但如果回调中嵌入了回调呢?如果层次很深，就是回调地狱。Java中的CompletableFuture其实就是Promise，用来解决回调地狱问题。Promise是为了让代码变得优美而存在的。

有多优美?这么说吧，一旦你使用了CompletableFuture，就会爱不释手，就像初恋女友一样，天天想着她。

一系列静态方法

从它的源代码中，我们可以看到，CompletableFuture直接提供了几个便捷的静态方法入口。其中有run和supply两组。

run的参数是Runnable，而supply的参数是Supplier。前者没有返回值，而后者有，否则没有什么两样。

这两组静态函数，都提供了传入自定义线程池的功能。如果你用的不是外置的线程池，那么它就会使用默认的ForkJoin线程池。默认的线程池，大小和用途你是控制不了的，所以还是建议自己传递一个。

典型的代码，写起来是这个样子。

 
 
 
 

  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(()->{ 
  
  
  
  
 return "test"; 
  
  
  
  
}); 
  
  
  
  
String result = future.join(); 
 
 
 
 

拿到CompletableFuture后，你就可以做更多的花样。

这些花样有很多

我们说面说了，CompletableFuture的主要作用，就是让代码写起来好看。配合Java8之后的stream流，可以把整个计算过程抽象成一个流。前面任务的计算结果，可以直接作为后面任务的输入，就像是管道一样。

 
 
 
 

  
  
  
  
thenApply 
  
  
  
  
thenApplyAsync 
  
  
  
  
thenAccept 
  
  
  
  
thenAcceptAsync 
  
  
  
  
thenRun 
  
  
  
  
thenRunAsync 
  
  
  
  
thenCombine 
  
  
  
  
thenCombineAsync 
  
  
  
  
thenCompose 
  
  
  
  
thenComposeAsync 
 
 
 
 

比如，下面代码的执行结果是99，并不因为是异步就打乱代码执行的顺序了。

 
 
 
 

  
  
  
  
CompletableFuture cf = CompletableFuture.supplyAsync(() -> 10) 
  
  
  
  
                .thenApplyAsync((e) -> { 
  
  
  
  
                    try { 
  
  
  
  
                        Thread.sleep(10000); 
  
  
  
  
                    } catch (InterruptedException ex) { 
  
  
  
  
                        ex.printStackTrace(); 
  
  
  
  
                    } 
  
  
  
  
                    return e * 10; 
  
  
  
  
                }).thenApplyAsync(e -> e - 1); 
  
  
  
  
 
  
  
  
  
cf.join(); 
  
  
  
  
System.out.println(cf.get()); 
 
 
 
 

同样的，函数的作用还要看then后面的动词。

• apply 有入参和返回值，入参为前置任务的输出
• accept 有入参无返回值，会返回CompletableFuture
• run 没有入参也没有返回值，同样会返回CompletableFuture
• combine 形成一个复合的结构，连接两个CompletableFuture，并将它们的2个输出结果，作为combine的输入
• compose 将嵌套的CompletableFuture平铺开，用来串联两个CompletableFuture

when和handle

上面的函数列表，其实还有很多。比如：

 
 
 
 

  
  
  
  
whenComplete 
 
 
 
 

when的意思，就是任务完成时候的回调。比如我们上面的例子，打算在完成任务后，输出一个done。它也是属于只有入参没有出参的范畴，适合放在最后一步进行观测。

 
 
 
 

  
  
  
  
CompletableFuture cf = CompletableFuture.supplyAsync(() -> 10) 
  
  
  
  
                .thenApplyAsync((e) -> { 
  
  
  
  
                    try { 
  
  
  
  
                        Thread.sleep(1000); 
  
  
  
  
                    } catch (InterruptedException ex) { 
  
  
  
  
                        ex.printStackTrace(); 
  
  
  
  
                    } 
  
  
  
  
                    return e * 10; 
  
  
  
  
                }).thenApplyAsync(e -> e - 1) 
  
  
  
  
                .whenComplete((r, e)->{ 
  
  
  
  
                    System.out.println("done"); 
  
  
  
  
                }) 
  
  
  
  
                ; 
  
  
  
  
 
  
  
  
  
cf.join(); 
  
  
  
  
System.out.println(cf.get()); 
 
 
 
 

handle和exceptionally的作用，和whenComplete是非常像的。

 
 
 
 

  
  
  
  
public CompletableFuture exceptionally(Function fn); 
  
  
  
  
 
  
  
  
  
public  CompletionStage handle(BiFunction fn); 
 
 
 
 

 
 
 
 

  
  
  
  
CompletableFuture cf = CompletableFuture.supplyAsync(() -> 10) 
  
  
  
  
                .thenApplyAsync(e->e/0) 
  
  
  
  
                .thenApplyAsync(e -> e - 1) 
  
  
  
  
                .exceptionally(ex->{ 
  
  
  
  
                    System.out.println(ex); 
  
  
  
  
                    return -1; 
  
  
  
  
                }); 
  
  
  
  
 
  
  
  
  
cf.join(); 
  
  
  
  
System.out.println(cf.get()); 
 
 
 
 

 
 
 
 

  
  
  
  
ExecutorService executor = Executors.newFixedThreadPool(5); 
  
  
  
  
 
  
  
  
  
CountDownLatch countDown = new CountDownLatch(requests.size()); 
  
  
  
  
for(Request request:requests){ 
  
  
  
  
    executor.execute(()->{ 
  
  
  
  
        try{ 
  
  
  
  
        //some opts 
  
  
  
  
        }finally{ 
  
  
  
  
            countDown.countDown(); 
  
  
  
  
        } 
  
  
  
  
    }); 
  
  
  
  
} 
  
  
  
  
countDown.await(200,TimeUnit.MILLISECONDS); 
 
 
 
 

 
 
 
 

  
  
  
  
ExecutorService executor = Executors.newFixedThreadPool(5); 
  
  
  
  
 
  
  
  
  
List> futureList = requests 
  
  
  
  
    .stream() 
  
  
  
  
    .map(request-> 
  
  
  
  
        CompletableFuture.supplyAsync(e->{ 
  
  
  
  
            //some opts 
  
  
  
  
        },executor)) 
  
  
  
  
    .collect(Collectors.toList()); 
  
  
  
  
 
  
  
  
  
CompletableFuture allCF = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])); 
  
  
  
  
 
  
  
  
  
allCF.join();