Java8系列 (七) CompletableFuture异步编程

2023-01-31 07:50:27

概述

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，截图如下：