CompleteFuture可以把一个复杂任务分解成由诸多计算节点组成的有向图。

三种常用关系

串行关系

并行关系

汇总关系

CompletableFuture 基本使用

创建异步任务

有返回的异步任务

@Test
    public void supplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> 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<Void> 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<String> audience = CompletableFuture.supplyAsync(() -> {
            System.out.println("正在洗菜");
            return "猪肉";
        });

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

        end.join();
    }

并行组合 + 汇总组合

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

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

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

        end.join();
    }

异常

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

        CompletableFuture<Void> 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<Object> handle = CompletableFuture.runAsync(() -> {
            throw new RuntimeException("异常");
        })
                .handleAsync((aVoid, throwable) -> {
                    try {
                        TimeUnit.SECONDS.sleep(3);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("我是handleAsync");
                    return null;
                })
                .handle((o, throwable) -> {
                    System.out.println("我是handle");
                    return null;
                })
                .handleAsync((avoid, throwable) -> {
                    System.out.println("我是handleAsync2");
                    throw new RuntimeException("处理异常中的异常");
//                  return null;
                })
                .exceptionally(throwable -> {
                    System.out.println("我是exceptionally1");
                    return null;
                })
                              .exceptionally(throwable -> {
                    System.out.println("我是exceptionally2");
                    return null;
                })
                ;

        handle.join();
    }

主动抛异常

/**
* 如果将主动抛异常的时机延长到6s, 
* 由于该节点计算5s就完成, 所以不会收到异常
*
* 否则收到异常:手动失败
*/
@Test
    public void obtException() throws InterruptedException, ExecutionException, TimeoutException {

        CompletableFuture<String> fu = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
                System.out.println("成功执行了");
            } catch (InterruptedException ignored) {
            }
            return null;
        });

        CompletableFuture<String> end = fu.exceptionally(throwable -> {
            System.out.println("收到异常:" + throwable.getMessage());
            return null;
        });


        TimeUnit.SECONDS.sleep(3);
                // 如果等6s后再抛，由于结果已经计算完成，则不会受到异常
                // TimeUnit.SECONDS.sleep(6);  
        fu.obtrudeException(new RuntimeException("手动失败"));
        end.join();
    }