­

Java8系列 (七) CompletableFuture异步编程

  • 2019 年 11 月 8 日
  • 筆記

概述

Java8之前用 Future 处理异步请求, 当你需要获取任务结果时, 通常的做法是调用  get(long timeout, TimeUnit unit) 此方法会阻塞当前的线程, 如果任务处理超时, 就会抛出一个  TimeoutException 

    @Test      public void test1() throws InterruptedException, ExecutionException, TimeoutException {          ExecutorService executorService = Executors.newCachedThreadPool();          Future<String> f = executorService.submit(() -> "ceshishanghu");          String s = f.get(3, TimeUnit.SECONDS);          System.out.println(s);      }

在Java8中引入了 CompletableFuture, 使用它提供的API可以不用像之前那样阻塞式或轮询的获取某个异步任务的结果, CompletableFuture 会在异步任务处理完成后自动进行回调, 让你可以链式的组合多个异步任务。

CompletableFuture 类中提供了许多以 Async 后缀结尾的方法。通常而言，名称中不带 Async 的方法和它的前一个任务一样，在同一个线程中运行。而名称以 Async 结尾的方法会将后续的任务提交到一个线程池，所以每个任务是由不同的线程处理的。

静态工厂方法

  • supplyAsync(): 异步处理任务, 有返回值
  • runAsync(): 异步处理任务, 没有返回值
  • allOf(): 需要等待所有的异步任务都执行完毕,才会返回一个新的CompletableFuture
  • anyOf(): 任意一个异步任务执行完毕,就会返回一个新的CompletableFuture
  • completedFuture(): 这种方式获取的 CompletableFuture 不是异步的，它会等待获取明确的返回结果之后再返回一个已经完成的 CompletableFuture
    @Test      public void test2() {          //创建一个已经有任务结果的CompletableFuture          CompletableFuture<String> f1 = CompletableFuture.completedFuture("return value");          //异步处理任务,有返回值          CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);          //异步处理任务,没有返回值          CompletableFuture<Void> f3 = CompletableFuture.runAsync(System.out::println);          //需要等待所有的异步任务都执行完毕,才会返回一个新的CompletableFuture  //        CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);          //任意一个异步任务执行完毕,就会返回一个新的CompletableFuture          CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2, f3);          Object result = any.join();          System.out.println("result = " + result);//result = return value      }        public String get() {          delay();          return "异步任务结果";      }        public void delay() {          try {              Thread.sleep(1000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }

上面的示例中, allOf() 因为要等待所有的异步任务执行完成，所以要延时1秒钟才会返回一个新的 CompletableFuture, 而 anyOf() 则不需要等待所有的异步任务, 因为第一个异步最先完成, 所以控制台输出  result = return value 。

链式调用

A任务执行完毕, 继续执行B任务, B任务执行完毕, 继续执行C任务…

    @Test      public void test2() {          CompletableFuture<Void> f = CompletableFuture.supplyAsync(() -> {              //测试抛异常后,handle()方法接受并处理  　　　　　　　//int x = 1 / 0;              return "这是一个栗子";          }).handle((res, ex) -> {              System.out.println("handle res = " + res);              if (Objects.nonNull(ex)) {                  System.out.println("handle ex" + ex.getCause().getMessage());              }              return Objects.nonNull(ex) ? 0 : 1;          }).thenApply(res -> {              System.out.println("thenApply res = " + res);              return res == 1 ? "success" : "error";          }).thenAccept(res -> System.out.println("thenAccept res = " + res)          ).thenRun(() -> System.out.println("没有参数, 异步执行一个没有返回值的任务"));          f.join();      }

输出结果:

handle res = 这是一个栗子  thenApply res = 1  thenAccept res = success  没有参数, 异步执行一个没有返回值的任务

将上面   int x = 1 / 0; 这行代码取消注释, 重新运行结果如下:

handle res = null  handle ex/ by zero  thenApply res = 0  thenAccept res = error  没有参数, 异步执行一个没有返回值的任务

可以看到, handle() 方法接受前一个 CompletableFuture  的返回结果或抛出的异常作为方法入参, 经过处理后再返回一个新的结果。

级联组合

  • thenCompose(): 对两个异步操作进行组合，第一个操作完成时，将其结果作为参数传递给第二个操作, 第二个操作会返回一个新的CompletableFuture。
  • thenCombine(): 将两个完全无关联的异步请求的结果整合起来, 计算出一个新的值并返回
    @Test      public void test3() {          CompletableFuture<String> f = CompletableFuture.completedFuture("CompletableFuture 1");          CompletableFuture<String> f1 = f.thenCompose(res -> {              System.out.println("thenCompose res = " + res);              return CompletableFuture.supplyAsync(() -> "CompletableFuture 2");          });          System.out.println(f1.join());          CompletableFuture<Integer> f3 = CompletableFuture.completedFuture(998);          CompletableFuture<String> f4 = f.thenCombine(f3, (str, num) -> {              System.out.println("str = " + str + ", num= " + num);              return str + num;          });          System.out.println(f4.join());      }

输出结果:

thenCompose res = CompletableFuture 1  CompletableFuture 2  str = CompletableFuture 1, num= 998  CompletableFuture 1998

whenComplete

当前一个 CompletableFuture  计算完成或抛出异常时, 可以使用 whenComplete() 执行指定的任务。

    @Test      public void test4() {          CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {              //测试抛异常后,whenComplete()方法接受并处理              int x = 1 / 0;              return "这是一个栗子";          }).whenComplete((res, ex) -> {              System.out.println("whenComplete res = " + res);              if (Objects.nonNull(ex)) {                  System.out.println("whenComplete ex" + ex.getCause().getMessage());              }          });          System.out.println("f.join() = " + f.join());      }

输出结果如下，其中 res 对应前一个 CompletableFuture 的返回结果，ex 对应前一个 CompletableFuture 抛出的异常(如果发生异常)。

从控制台输出顺序看出，当前一个 CompletableFuture  计算完成或抛出异常时,  whenComplete() 会接受它的返回结果或抛出的异常，来做一些其他的事情，最后再返回原来的返回结果或抛出异常。类比下 try/catch 语句块中的 final 语句块。

whenComplete res = null  whenComplete ex/ by zero    java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero        at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273)      at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280)      at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1592)      at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java)      at java.util.concurrent.CompletableFuture$AsyncSupply.exec(CompletableFuture.java:1582)      at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)      at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)      at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)      at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)  Caused by: java.lang.ArithmeticException: / by zero      at com.java8.action.ChapterTest.lambda$test4$0(ChapterTest.java:22)

异常处理

只有当前一个 CompletableFuture 发生异常时，才会进入到 exceptionally() 方法，并将产生的异常作为入参。

    @Test      public void test5() {          CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {              //测试抛异常后,exceptionally()方法接受并处理              //int x = 1 / 0;              return "这是一个栗子";          }).exceptionally(ex -> ex.getCause().getMessage());          System.out.println("f.join() = " + f.join());      }

注释  int x = 1 / 0; ，输出如下：

f.join() = 这是一个栗子

取消注释   int x = 1 / 0; , 输出如下：

f.join() = / by zero

Both系列方法

  • thenAcceptBoth(): 等待当前的 CompletableFuture 和另一个 CompletableFuture 执行完成，将它们的返回结果作为入参去执行一个操作，没有返回值
  • runAfterBoth(): 等待当前的 CompletableFuture 和另一个 CompletableFuture 执行完成，然后去执行一个操作，没有返回值

代码清单一

    @Test      public void test6() {          CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);          CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);          CompletableFuture<Void> both = f1.thenAcceptBoth(f2, (num, str) -> System.out.println("num = " + num + ", str = " + str));          both.join();      }        public String get() {          delay();          return "CompletableFuture 2";      }        public void delay() {          try {              Thread.sleep(1000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }

代码清单一输出结果如下：

num = 9523, str = CompletableFuture 2 

代码清单二

    @Test      public void test7() {          CompletableFuture<Integer> f1 = CompletableFuture.completedFuture(9523);          CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "");          CompletableFuture<Void> both = f1.runAfterBoth(f2, () -> System.out.println("执行一个任务，没有入参"));          both.join();      }

代码清单二输出结果如下：

执行一个任务，没有入参

Either系列

  • acceptEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，将对应的返回结果作为入参去执行一个操作，没有返回值
  • applyToEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，将对应的返回结果作为入参，使用 mapping 函数转换成一个新的值并返回
  • runAfterEither: 当前的 CompletableFuture 和另一个 CompletableFuture 任意一个执行完成，然后去执行一个操作，没有返回值

代码清单三：

    @Test      public void test8() {          CompletableFuture<String> f1 = CompletableFuture.completedFuture("CompletableFuture 1");          CompletableFuture<String> f2 = CompletableFuture.supplyAsync(this::get);          CompletableFuture<Void> both = f1.acceptEither(f2, System.out::println);          both.join();      }        public String get() {          delay();          return "CompletableFuture 2";      }        public void delay() {          try {              Thread.sleep(1000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }

代码清单三输出结果：

CompletableFuture 1

代码清单四：

    @Test      public void test9() {          CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);          CompletableFuture<String> f2 = CompletableFuture.completedFuture("CompletableFuture 2");          CompletableFuture<Integer> f3 = f1.applyToEither(f2, res -> {              System.out.println("res = " + res);              return res.length();          });          System.out.println("f3.join() = " + f3.join());      }        public String get() {          delay();//这里会延时一秒钟          return "CompletableFuture 1";      }

代码清单四输出结果：

res = CompletableFuture 2  f3.join() = 19

代码清单五：

    @Test      public void test10() {          CompletableFuture<String> f1 = CompletableFuture.supplyAsync(this::get);          CompletableFuture<Void> f2 = CompletableFuture.allOf();          CompletableFuture<Void> f3 = f1.runAfterEither(f2, () -> System.out.println("执行一个任务，没有入参"));          f3.join();      }        public String get() {          delay();//这里会延时一秒钟          return "CompletableFuture 1";      }

代码清单五输出结果：

执行一个任务，没有入参

使用自定义的执行器来处理多个异步任务

在实际应用场景中可能会遇到这种情况，假如你需要同时处理大量的异步任务，且这些异步任务互相不依赖，你只要最后把它们的结果组装起来就行，这该怎么实现呢？

下面给出了一个使用默认执行器的示例，通过Stream流同时创建 9 个异步任务，获取它们的结果并组装后返回，其中 Runtime.getRuntime().availableProcessors() 表示Java虚拟机可用的处理器个数，在我之前的文章 Java8系列 (二) Stream流 中有介绍过。

代码清单六：

    @Test      public void test11() {          List<String> list = Arrays.asList("王小波书店", "杭州沈记古旧书店", "猫的天空之城概念书店", "纯真年代书吧", "南山书屋", "西西弗书店", "新华书店", "钟书阁", "云门书屋");          System.out.println("当前机器有" + Runtime.getRuntime().availableProcessors() + "个可用的处理器");          long start = System.nanoTime();          List<CompletableFuture<String>> futures = list.stream()                  .map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str)))                  .collect(Collectors.toList());          System.out.println("get futures "+(System.nanoTime() - start) / 1000_000 + " msecs");          String result = futures.stream()                  .map(CompletableFuture::join)                  .collect(Collectors.joining(",", "[", "]"));          System.out.println("get result "+(System.nanoTime() - start) / 1000_000 + " msecs");          System.out.println(result);      }        public String calculateLength(String str) {          delay();          return str;      }        public void delay() {          try {              Thread.sleep(1000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }

运行代码清单六，输出结果：

当前机器有4个可用的处理器  get futures 95 msecs  get result 3098 msecs  [王小波书店,杭州沈记古旧书店,猫的天空之城概念书店,纯真年代书吧,南山书屋,西西弗书店,新华书店,钟书阁,云门书屋]

可以看到，虽然使用了异步处理，但还是花了 3098 毫秒才执行完成所有任务。这是因为 CompletableFuture 内部采用的是通用线程池 ForkJoinPool.commonPool() , 默认都使用固定数目的线程, 具体线程数取决于  Runtime.getRuntime().availableProcessors()  的返回值。

我这里测试的机器显示通用线程池中处于可用状态的线程数为 4，一次只能同时处理 4 个任务，后面的5个异步任务只能等到前面某一个操作完成释放出空闲线程才能继续, 因此总的会消耗约 3 秒钟的时间。

我们将上面的代码进行重构，使用自定义的执行器，通过自定义的执行器你可以指定线程池的大小。其中线程数的设定可以参考公式  Nthreads = NCPU * UCPU * (1 + W/C) 

    @Test      public void test12() {          List<String> list = Arrays.asList("王小波书店", "杭州沈记古旧书店", "猫的天空之城概念书店", "纯真年代书吧", "南山书屋", "西西弗书店", "新华书店", "钟书阁", "云门书屋");          final ExecutorService executor = Executors.newFixedThreadPool(Math.min(list.size(), 100), r -> {              Thread thread = new Thread(r);              //守护线程不会组织程序的终止              thread.setDaemon(true);              return thread;          });          System.out.println("当前机器有" + Runtime.getRuntime().availableProcessors() + "个可用的处理器, 当前处理异步请求的线程池大小为 " + Math.min(list.size(), 100));          long start = System.nanoTime();          List<CompletableFuture<String>> futures = list.stream()                  .map(str -> CompletableFuture.supplyAsync(() -> this.calculateLength(str), executor))                  .collect(Collectors.toList());          System.out.println("get futures " + (System.nanoTime() - start) / 1000_000 + " msecs");          String result = futures.stream()                  .map(CompletableFuture::join)                  .collect(Collectors.joining(",", "[", "]"));          System.out.println("get result " + (System.nanoTime() - start) / 1000_000 + " msecs");          System.out.println(result);      }        public String calculateLength(String str) {          delay();          return str;      }        public void delay() {          try {              Thread.sleep(1000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }

输出结果如下：

当前机器有4个可用的处理器, 当前处理异步请求的线程池大小为 9  get futures 38 msecs  get result 1039 msecs  [王小波书店,杭州沈记古旧书店,猫的天空之城概念书店,纯真年代书吧,南山书屋,西西弗书店,新华书店,钟书阁,云门书屋]

可以看到，使用自定义的执行器调大线程池大小后，总的运行时间只要 1039 毫秒。

将CompletableFuture作为Controller的返回值

上面还存在一个问题，虽然现在可以同时处理多个异步任务，但是如果需要将异步结果返回给另一个服务，那不是还得通过 join() 阻塞的获取到返回值后才能再返回么？

自Spring Boot 1.3 (Spring 4.2) 之后开始支持 CompletableFuture 或 CompletionStage 作为 Controller 的返回值，她很好的解决了上面的异步阻塞问题，只要将  CompletableFuture 作为 Controller 的返回值，在异步任务执行完成后，它会自动响应结果给另一个服务。

@RestController  public class AsyncController {        @GetMapping("/redirect")      public CompletableFuture<ModelAndView> redirect() {          return CompletableFuture.supplyAsync(() -> {              this.delay();              RedirectView redirectView = new RedirectView("https://www.cnblogs.com/qingshanli/");              redirectView.addStaticAttribute("hint", "CompletableFuture组装ModelAndView视图,异步返回结果");              return new ModelAndView(redirectView);          });      }        @GetMapping("/async")      public CompletableFuture<String> async() {          System.out.println("async method start");          return CompletableFuture.supplyAsync(() -> {              this.delay();              return "CompletableFuture作为Controller的返回值,异步返回结果";          }).whenComplete((res, ex) -> System.out.println("async method completely, res = " + res + ", ex = " + ex));      }        public void delay() {          try {              Thread.sleep(3000L);          } catch (InterruptedException e) {              e.printStackTrace();          }      }  } 

启动项目，Postman 访问 http://localhost:8080/async，截图如下：

Postman 访问 http://localhost:8080/redirect，截图如下：

参考资料

https://github.com/AndreasKl/spring-boot-mvc-completablefuture

https://nickebbitt.github.io/blog/2017/03/22/async-web-service-using-completable-future

https://www.humansreadcode.com/spring-boot-completablefuture/

Java8 实战

作者：张小凡
出处：https://www.cnblogs.com/qingshanli/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。如果觉得还有帮助的话，可以点一下右下角的【推荐】。

VirMach 便宜 VPS

QNews

QNews