实现异步编程，这个工具类你得掌握！

 [[393773]]

本文转载自微信公众号「月伴飞鱼」，作者日常加油站。转载本文请联系月伴飞鱼公众号。

我们提供的服务有：成都网站设计、成都网站制作、外贸网站建设、微信公众号开发、网站优化、网站认证、韶山ssl等。为上千企事业单位解决了网站和推广的问题。提供周到的售前咨询和贴心的售后服务，是有科学管理、有技术的韶山网站制作公司

前言

最近看公司代码，多线程编程用的比较多，其中有对CompletableFuture的使用，所以想写篇文章总结下

在日常的Java8项目开发中，CompletableFuture是很强大的并行开发工具，其语法贴近java8的语法风格，与stream一起使用也能大大增加代码的简洁性

大家可以多应用到工作中，提升接口性能，优化代码

基本介绍

CompletableFuture是Java 8新增的一个类，用于异步编程，继承了Future和CompletionStage

这个Future主要具备对请求结果独立处理的功能，CompletionStage用于实现流式处理，实现异步请求的各个阶段组合或链式处理，因此completableFuture能实现整个异步调用接口的扁平化和流式处理，解决原有Future处理一系列链式异步请求时的复杂编码

Future的局限性

1、Future 的结果在非阻塞的情况下，不能执行更进一步的操作

我们知道，使用Future时只能通过isDone()方法判断任务是否完成，或者通过get()方法阻塞线程等待结果返回，它不能非阻塞的情况下，执行更进一步的操作。

2、不能组合多个Future的结果

假设你有多个Future异步任务，你希望最快的任务执行完时，或者所有任务都执行完后，进行一些其他操作

3、多个Future不能组成链式调用

当异步任务之间有依赖关系时，Future不能将一个任务的结果传给另一个异步任务，多个Future无法创建链式的工作流。

4、没有异常处理

现在使用CompletableFuture能帮助我们完成上面的事情，让我们编写更强大、更优雅的异步程序

基本使用

创建异步任务

通常可以使用下面几个CompletableFuture的静态方法创建一个异步任务

 
 
 
 

  
  
  
  
public static CompletableFuture runAsync(Runnable runnable);              //创建无返回值的异步任务
  
  
  
  
public static CompletableFuture runAsync(Runnable runnable, Executor executor);     //无返回值，可指定线程池（默认使用ForkJoinPool.commonPool）
  
  
  
  
public static  CompletableFuture supplyAsync(Supplier supplier);           //创建有返回值的异步任务

  
  
  
  
public static  CompletableFuture supplyAsync(Supplier supplier, Executor executor); //有返回值，可指定线程池

 
 
 
 

使用示例：

 
 
 
 

  
  
  
  
Executor executor = Executors.newFixedThreadPool(10);
  
  
  
  
CompletableFuture future = CompletableFuture.runAsync(() -> {
  
  
  
  
    //do something
  
  
  
  
}, executor);
  
  
  
  
int poiId = 111;
  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
  
  
  
  
 PoiDTO poi = poiService.loadById(poiId);
  
  
  
  
  return poi.getName();
  
  
  
  
});
  
  
  
  
// Block and get the result of the Future
  
  
  
  
String poiName = future.get();
 
 
 
 

使用回调方法

通过future.get()方法获取异步任务的结果，还是会阻塞的等待任务完成

CompletableFuture提供了几个回调方法，可以不阻塞主线程，在异步任务完成后自动执行回调方法中的代码

 
 
 
 

  
  
  
  
public CompletableFuture thenRun(Runnable runnable);            //无参数、无返回值
  
  
  
  
public CompletableFuture thenAccept(Consumer action);         //接受参数，无返回值
  
  
  
  
public  CompletableFuture thenApply(Function fn); //接受参数T，有返回值U

 
 
 
 

使用示例：

 
 
 
 

  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
  
  
  
  
                           .thenRun(() -> System.out.println("do other things. 比如异步打印日志或发送消息"));
  
  
  
  
//如果只想在一个CompletableFuture任务执行完后，进行一些后续的处理，不需要返回值，那么可以用thenRun回调方法来完成。
  
  
  
  
//如果主线程不依赖thenRun中的代码执行完成，也不需要使用get()方法阻塞主线程。
 
 
 
 

 
 
 
 

  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
  
  
  
  
                           .thenAccept((s) -> System.out.println(s + " world"));
  
  
  
  
//输出：Hello world
  
  
  
  
//回调方法希望使用异步任务的结果，并不需要返回值，那么可以使用thenAccept方法
 
 
 
 

 
 
 
 

  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(() -> {
  
  
  
  
  PoiDTO poi = poiService.loadById(poiId);
  
  
  
  
  return poi.getMainCategory();
  
  
  
  
}).thenApply((s) -> isMainPoi(s));   // boolean isMainPoi(int poiId);
  
  
  
  

  
  
  
  
future.get();
  
  
  
  
//希望将异步任务的结果做进一步处理，并需要返回值，则使用thenApply方法。
  
  
  
  
//如果主线程要获取回调方法的返回，还是要用get()方法阻塞得到
 
 
 
 

组合两个异步任务

 
 
 
 

  
  
  
  
//thenCompose方法中的异步任务依赖调用该方法的异步任务
  
  
  
  
public  CompletableFuture thenCompose(Function> fn); 

  
  
  
  
//用于两个独立的异步任务都完成的时候
  
  
  
  
public  CompletableFuture thenCombine(CompletionStage other, 
  
  
  
  
                                              BiFunction fn); 
 
 
 
 

使用示例：

 
 
 
 

  
  
  
  
CompletableFuture> poiFuture = CompletableFuture.supplyAsync(
  
  
  
  
  () -> poiService.queryPoiIds(cityId, poiId)
  
  
  
  
);
  
  
  
  
//第二个任务是返回CompletableFuture的异步方法
  
  
  
  
CompletableFuture> getDeal(List poiIds){
  
  
  
  
  return CompletableFuture.supplyAsync(() ->  poiService.queryPoiIds(poiIds));
  
  
  
  
}
  
  
  
  
//thenCompose
  
  
  
  
CompletableFuture> resultFuture = poiFuture.thenCompose(poiIds -> getDeal(poiIds));
  
  
  
  
resultFuture.get();
 
 
 
 

thenCompose和thenApply的功能类似，两者区别在于thenCompose接受一个返回CompletableFuture的Function，当想从回调方法返回的CompletableFuture中直接获取结果U时，就用thenCompose

如果使用thenApply，返回结果resultFuture的类型是CompletableFuture>>，而不是CompletableFuture>

 
 
 
 

  
  
  
  
CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
  
  
  
  
  .thenCombine(CompletableFuture.supplyAsync(() -> "world"), (s1, s2) -> s1 + s2);
  
  
  
  
//future.get()
 
 
 
 

组合多个CompletableFuture

当需要多个异步任务都完成时，再进行后续处理，可以使用allOf方法

 
 
 
 

  
  
  
  
CompletableFuture poiIDTOFuture = CompletableFuture
  
  
  
  
 .supplyAsync(() -> poiService.loadPoi(poiId))
  
  
  
  
  .thenAccept(poi -> {
  
  
  
  
    model.setModelTitle(poi.getShopName());
  
  
  
  
    //do more thing
  
  
  
  
  });
  
  
  
  

  
  
  
  
CompletableFuture productFuture = CompletableFuture
  
  
  
  
 .supplyAsync(() -> productService.findAllByPoiIdOrderByUpdateTimeDesc(poiId))
  
  
  
  
  .thenAccept(list -> {
  
  
  
  
    model.setDefaultCount(list.size());
  
  
  
  
    model.setMoreDesc("more");
  
  
  
  
  });
  
  
  
  
//future3等更多异步任务，这里就不一一写出来了
  
  
  
  

  
  
  
  
CompletableFuture.allOf(poiIDTOFuture, productFuture, future3, ...).join();  //allOf组合所有异步任务，并使用join获取结果
 
 
 
 

该方法挺适合C端的业务，比如通过poiId异步的从多个服务拿门店信息，然后组装成自己需要的模型，最后所有门店信息都填充完后返回

这里使用了join方法获取结果，它和get方法一样阻塞的等待任务完成

多个异步任务有任意一个完成时就返回结果，可以使用anyOf方法

 
 
 
 

  
  
  
  
CompletableFuture future1 = CompletableFuture.supplyAsync(() -> {
  
  
  
  
    try {
  
  
  
  
        TimeUnit.SECONDS.sleep(2);
  
  
  
  
    } catch (InterruptedException e) {
  
  
  
  
       throw new IllegalStateException(e);
  
  
  
  
    }
  
  
  
  
    return "Result of Future 1";
  
  
  
  
});
  
  
  
  

  
  
  
  
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
  
  
  
  
    try {
  
  
  
  
        TimeUnit.SECONDS.sleep(1);
  
  
  
  
    } catch (InterruptedException e) {
  
  
  
  
       throw new IllegalStateException(e);
  
  
  
  
    }
  
  
  
  
    return "Result of Future 2";
  
  
  
  
});
  
  
  
  

  
  
  
  
CompletableFuture future3 = CompletableFuture.supplyAsync(() -> {
  
  
  
  
    try {
  
  
  
  
        TimeUnit.SECONDS.sleep(3);
  
  
  
  
    } catch (InterruptedException e) {
  
  
  
  
       throw new IllegalStateException(e);
  
  
  
  
      return "Result of Future 3";
  
  
  
  
});
  
  
  
  

  
  
  
  
CompletableFuture