Java CompleteFuture 书写逻辑清晰的异步任务

Future 是异步计算的结果表示，与 NodeJS 中的 Promise 类似。继承了两种能力：异步任务和流水线处理。

类图

三种常用关系

串行关系

并行关系

汇总关系

CompletableFuture 基本使用

创建异步任务

有返回的异步任务

@Test
    public void supplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture boil = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                return "烧水";
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        String s = boil.get();
        Assert.assertEquals("烧水", s);
    }

没有返回的异步任务

@Test
    public void runAsync() {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        CompletableFuture run = CompletableFuture.runAsync(() -> {

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        run.join();
        stopWatch.stop();
        double totalTimeSeconds = stopWatch.getTotalTimeSeconds();
        Assert.assertEquals(1, totalTimeSeconds, 0.1);
    }

串行组合

/**
* 正在洗菜
*切菜:猪肉
*/
@Test
    public void thenAccept() {
        CompletableFuture audience = CompletableFuture.supplyAsync(() -> {
            System.out.println("正在洗菜");
            return "猪肉";
        });

        CompletableFuture end = audience.thenAccept(it -> {
            System.out.println("切菜:" + it);
        });

        end.join();
    }

并行组合 + 汇总组合

/**
*正在烧水...
*正在包饺子
*水开了
*饺子包好了
*准备，下10个饺子
*/
@Test
    public void thenAcceptBoth() {
        CompletableFuture water = CompletableFuture.runAsync(() -> {
            try {
                System.out.println("正在烧水...");
                TimeUnit.SECONDS.sleep(4);
                System.out.println("水开了");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        CompletableFuture dumpling = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println("正在包饺子");
                TimeUnit.SECONDS.sleep(5);
                System.out.println("饺子包好了");
                return "10个饺子";
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return null;
        });

        CompletableFuture end = water.thenAcceptBoth(dumpling, (a, b) -> {
            System.out.println("准备，下" + b);
        });

        end.join();
    }

异常

/**
*正在烧水...
*充钱
*余额不足
*/
@Test
    public void exception() {
        CompletableFuture water = CompletableFuture.runAsync(() -> {
                System.out.println("正在烧水...");
                TimeUnit.SECONDS.sleep(4);
                throw new IllegalArgumentException("没燃气了");     
        });

        CompletableFuture end = water.exceptionally(throwable -> {
            System.out.println("充钱");
            throw new RuntimeException("余额不足");
        });
        try {
            end.join();
        } catch (CompletionException e) {
            System.out.println(e.getCause().getMessage());
            Assert.assertEquals("余额不足", e.getCause().getMessage());
        }
    }

异常的抓获与处理

/*
* 1. handle..() 不会抓获异常 ，所以配置多个都会被执行;
* 2. exceptionally() 会抓获异常所以只生效一次。
* 3. exceptionally() 必须在结束前调用，否则不生效

*我是handleAsync
*我是handleAsync2
*我是handle
*我是exceptionally1
*/
    @Test
    public void multiExceptionHandle() {
        CompletableFuture