异步编程 - 05 基于JDK中的Future实现异步编程(中)_CompletableFuture

// 0.创建线程池
private static final ThreadPoolExecutor bizPoolExecutor = new ThreadPoolExecutor(8, 8, 1, TimeUnit.MINUTES,
        new LinkedBlockingQueue<>(10));

// 2. 有返回值的异步执行
public static void supplyAsyncWithBizExecutor() throws InterruptedException, ExecutionException {
    // 2.1创建异步任务，并返回future
    CompletableFuture future = CompletableFuture.supplyAsync(new Supplier<String>() {
        @Override
        public String get() {
            // 2.1.1休眠2s模拟任务计算
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            // 2.1.2 返回异步计算结果
            return "hello,jiaduo";
        }
    }, bizPoolExecutor);

    // 2.2 同步等待异步任务执行结束
    System.out.println(future.get());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

3）基于thenRun实现异步任务A，执行完毕后，激活异步任务B执行 (需要注意的是，这种方式激活的异步任务B是拿不到任务A的执行结果的)

需要注意的是，这种方式激活的异步任务B是拿不到任务A的执行结果的：

    // I thenRun不能访问oneFuture的结果
    public static void thenRun() throws InterruptedException, ExecutionException {
        // 1.创建异步任务，并返回future
        CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {
                // 1.1休眠2s，模拟任务计算
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 1.2返回计算结果
                return "hello";
            }
        });
        // 2.在future上施加事件，当future计算完成后回调该事件，并返回新future
        CompletableFuture twoFuture = oneFuture.thenRun(new Runnable() {

            @Override
            public void run() {
                // 2.1.1当oneFuture任务计算完成后做一件事情
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName());
                System.out.println("---after oneFuture over doSomething---");
            }
        });

        // 3.同步等待twoFuture对应的任务完成，返回结果固定为null
        System.out.println(twoFuture.get());

    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

由上述代码可知，代码1创建异步任务，并返回oneFuture对象，代码2在oneFuture上调用thenRun方法添加异步执行事件，当oneFuture计算完成后回调该事件，并返回twoFuture，另外，在twoFuture上调用get()方法也会返回null，因为回调事件是没有返回值的。

默认情况下oneFuture对应的异步任务和在oneFuture上添加的回调事件都是使用ForkJoinPool.commonPool()中的同一个线程来执行的，大家可以使用thenApplyAsync(Function fn, Executor executor)来指定设置的回调事件使用自定义线程池线程来执行，也就是oneFuture对应的任务与在其上设置的回调执行将不会在同一个线程中执行。

4）基于thenAccept实现异步任务A，执行完毕后，激活异步任务B执行 (需要注意的是，这种方式激活的异步任务B是可以拿到任务A的执行结果的)

需要注意的是，这种方式激活的异步任务B是可以拿到任务A的执行结果的：

public static void thenAccept() throws InterruptedException, ExecutionException {
    // 1.创建异步任务，并返回future
    CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

        @Override
        public String get() {

            // 1.1休眠2s，模拟任务计算
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            // 1.2返回计算结果
            return "hello";
        }
    });
    // 2.在future上施加事件，当future计算完成后回调该事件，并返回新future
    CompletableFuture twoFuture = oneFuture.thenAccept(new Consumer<String>() {

        @Override
        public void accept(String t) {
            // 2.1.1对oneFuture返回的结果进行加工
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }

            System.out.println("---after oneFuture over doSomething---" + t);
        }
    });

    // 3.同步等待twoFuture对应的任务完成，返回结果固定为null
    System.out.println(twoFuture.get());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

在上述代码中，代码1创建异步任务，并返回oneFuture，代码2在oneFuture上调用thenAccept添加了一个任务，这个任务会在oneFuture对应的任务执行完毕后被激活执行。

需要注意的是，这里可以在回调的方法accept(String t)的参数t中来获取oneFuture对应的任务结果，另外需要注意的是，由于accept(String t)方法没有返回值，所以在twoFuture上调用get()方法最终也会返回null。

在默认情况下，oneFuture对应的异步任务和在oneFuture上添加的回调事件都是使用ForkJoinPool.commonPool()中的同一个线程来执行的，大家可以使用thenAccept-Async(Consumeraction，Executor executor)来指定设置的回调事件使用自定义线程池线程来执行，也就是oneFuture对应的任务与在其上设置的回调执行将不会在同一个线程中执行。

5）基于thenApply实现异步任务A，执行完毕后，激活异步任务B执行 (需要注意的是，这种方式激活的异步任务B是可以拿到任务A的执行结果的，并且可以获取到异步任务B的执行结果)

需要注意的是，这种方式激活的异步任务B是可以拿到任务A的执行结果的，并且可以获取到异步任务B的执行结果

public class TestCompletableFutureCallBack {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 1.创建异步任务，并返回future
        CompletableFuture<String> oneFuture = CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {

                // 1.1休眠2s，模拟任务计算
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 1.2返回计算结果
                return "hello";
            }
        });

        // 2.在future上施加事件，当future计算完成后回调该事件，并返回新future
        CompletableFuture<String> twoFuture = oneFuture.thenApply(new Function<String, String>() {

            // 2.1在步骤1计算结果基础上进行计算，这里t为步骤1返回的hello
            @Override
            public String apply(String t) {
                // 2.1.1对oneFuture返回的结果进行加工
                try {
                    Thread.sleep(1000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                // 2.1.2返回加工后结果
                return t + " artisan";
            }
        });

        // 3.同步等待twoFuture对应的任务完成，并获取结果
        System.out.println(twoFuture.get());

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

在上述代码中，代码1创建异步任务，并返回oneFuture，代码2在oneFuture上调用thenApply添加了一个任务，这个任务会在oneFuture对应的任务执行完毕后被激活执行。需要注意的是，这里可以在回调方法apply(String t)的参数t中获取oneFuture对应的任务结果，另外需要注意的是，由于apply(String t)方法有返回值，所以在twoFuture上调用get()方法最终也会返回回调方法返回的值。

默认情况下oneFuture对应的异步任务和在oneFuture上添加的回调事件都是使用ForkJoinPool.commonPool()中的同一个线程来执行的，大家可以使用thenApplyAsync(Functionfn，Executor executor)来指定设置的回调事件使用自定义线程池线程来执行，也就是oneFuture对应的任务与在其上设置的回调执行将不会在同一个线程中执行。

6）基于whenComplete设置回调函数，当异步任务执行完毕后进行回调，不会阻塞调用线程

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {

    // 1.创建一个CompletableFuture对象
    CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {

        @Override
        public String get() {
            // 1.1模拟异步任务执行
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 1.2返回计算结果
            return "hello,jiaduo";
        }
    });

    // 2.添加回调函数
    future.whenComplete(new BiConsumer<String, Throwable>() {

        @Override
        public void accept(String t, Throwable u) {
            // 2.1如果没有异常，打印异步任务结果
            if (null == u) {
                System.out.println(t);
            } else {
                // 2.2打印异常信息
                System.out.println(u.getLocalizedMessage());

            }
        }
    });

    // 3.挂起当前线程，等待异步任务执行完毕
    Thread.currentThread().join();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

这里代码1开启了一个异步任务，任务内先休眠1s，然后代码1.2返回计算结果；代码2则在返回的future上调用whenComplete设置一个回调函数，然后main线程就返回了。

在整个异步任务的执行过程中，main函数所在线程是不会被阻塞的，等异步任务执行完毕后会回调设置的回调函数，在回调函数内，代码2.1表示如果发现异步任务执行正常则打印执行结果，否则打印异常信息。这里代码3挂起了main函数所在线程，是因为具体执行异步任务的是ForkJoin的commonPool线程池，其中线程都是Deamon线程，所以，当唯一的用户线程main线程退出后整个JVM进程就退出了，会导致异步任务得不到执行。

小结

如上所述，当我们使用CompletableFuture实现异步编程时，大多数时候是不需要显式创建线程池，并投递任务到线程池内的。

我们只需要简单地调用CompletableFuture的runAsync或者supplyAsync等方法把异步任务作为参数即可，其内部会使用ForkJoinPool线程池来进行异步执行的支持，这大大简化了我们异步编程的负担，实现了声明式编程（告诉程序我要执行异步任务，但是具体怎么实现我不需要管），当然如果你想使用自己的线程池来执行任务，也是可以非常方便地进行设置的。

多个CompletableFuture进行组合运算

CompletableFuture功能强大的原因之一是其可以让两个或者多个Completable-Future进行运算来产生结果，下面我们来看其提供的几组函数：

1）基于thenCompose实现当一个CompletableFuture执行完毕后，执行另外一个CompletableFuture：

public class TestTwoCompletableFuture {
    // 1.异步任务，返回future
    public static CompletableFuture<String> doSomethingOne(String encodedCompanyId) {
        // 1.1创建异步任务
        return CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {

                // 1.1.1休眠1s，模拟任务计算
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                // 1.1.2 解密，并返回结果
                String id = encodedCompanyId;
                return id;
            }
        });
    }

    // 2.开启异步任务，返回future
    public static CompletableFuture<String> doSomethingTwo(String companyId) {
        return CompletableFuture.supplyAsync(new Supplier<String>() {

            @Override
            public String get() {

                // 2.1 休眠3s，模拟计算
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                // 2.2 查询公司信息，转换为str，并返回
                String str = companyId + ":alibaba";
                return str;
            }
        });
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // I，等doSomethingOne执行完毕后，接着执行doSomethingTwo
        CompletableFuture result = doSomethingOne("123").thenCompose(id -> doSomethingTwo(id));
        System.out.println(result.get());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

上述main函数中首先调用方法doSomethingOne(“123”)开启了一个异步任务，并返回了对应的CompletableFuture对象，我们取名为future1，然后在future1的基础上调用了thenCompose方法，企图让future1执行完毕后，激活使用其结果作为doSomethingTwo(String companyId)方法的参数的任务。

2）基于thenCombine实现当两个并发运行的CompletableFuture任务都完成后，使用两者的结果作为参数再执行一个异步任务，这里只需要把上面例子中的：

CompletableFuture result = doSomethingOne("123").thenCompose(id -> doSomethingTwo(id));
1

修改为：

result = doSomethingOne("123").thenCombine(doSomethingTwo("456"), (one, two) -> {
            return one + " " + two;
        });
1
2
3

3）基于allOf等待多个并发运行的CompletableFuture任务执行完毕：

public static void allOf() throws InterruptedException, ExecutionException {
    // 1.创建future列表
    List<CompletableFuture<String>> futureList = new ArrayList<>();
    futureList.add(doSomethingOne("1"));
    futureList.add(doSomethingOne("2"));
    futureList.add(doSomethingOne("3"));
    futureList.add(doSomethingOne("4"));

    // 2.转换多个future为一个
    CompletableFuture<Void> result = CompletableFuture
            .allOf(futureList.toArray(new CompletableFuture[futureList.size()]));

    // 3.等待所有future都完成
    System.out.println(result.get());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

如上代码1调用了四次doSomethingOne方法，分别返回一个CompletableFuture对象，然后收集这些CompletableFuture到futureList列表。

代码2调用allOf方法把多个CompletableFuture转换为一个result，代码3在result上调用get()方法会阻塞调用线程，直到futureList列表中所有任务执行完毕才返回。

4）基于anyOf等多个并发运行的CompletableFuture任务中有一个执行完毕就返回

public static void anyOf() throws InterruptedException, ExecutionException {
    // 1.创建future列表
    List<CompletableFuture<String>> futureList = new ArrayList<>();
    futureList.add(doSomethingOne("1"));
    futureList.add(doSomethingOne("2"));
    futureList.add(doSomethingTwo("3"));

    // 2.转换多个future为一个
    CompletableFuture<Object> result = CompletableFuture
            .anyOf(futureList.toArray(new CompletableFuture[futureList.size()]));

    // 3.等待某一个future完成
    System.out.println(result.get());

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

如上代码1调用了四次doSomethingOne方法，分别返回一个CompletableFuture对象，然后收集这些CompletableFuture到futureList列表。

代码2调用anyOf方法把多个CompletableFuture转换为一个result，代码3在result上调用get()方法会阻塞调用线程，直到futureList列表中有一个任务执行完毕才返回。

异常处理

前文的代码为我们演示的功能都是当异步任务内可以正常设置任务结果时的情况，但是情况并不总是这样的，比如下面这段代码：

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
    // 1.创建一个CompletableFuture对象
    CompletableFuture<String> future = new CompletableFuture<String>();

    // 2.开启线程计算任务结果，并设置
    new Thread(() -> {

        // 2.1休眠3s，模拟任务计算
        try {
            // 2.1.1抛出异常
            if (true) {
                throw new RuntimeException("excetion test");
            }
            // 2.1.2设置正常结果
            future.complete("ok");
        } catch (Exception e) {

        }
        // 2.2设置计算结果到future
        System.out.println("----" + Thread.currentThread().getName() + " set future result----");

    }, "thread-1").start();

    // 3.等待计算结果
    System.out.println(future.get());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

由上述代码可知，在代码2.1.2设置正常结果前，代码2.1.1抛出了异常，这会导致代码3一直阻塞，所以我们不仅需要考虑正常设置结果的情况，还需要考虑异常的情况，其实CompletableFuture提供了completeExceptionally方法来处理异常情况，将上述代码修改为如下所示。

public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
    // 1.创建一个CompletableFuture对象
    CompletableFuture<String> future = new CompletableFuture<String>();

    // 2.开启线程计算任务结果，并设置
    new Thread(() -> {

        // 2.1休眠3s，模拟任务计算
        try {
            // 2.1.1 抛出异常
            if (true) {
                throw new RuntimeException("excetion test");
            }
            // 2.1.2设置正常结果
            future.complete("ok");
        } catch (Exception e) {
            // 2.1.3 设置异常结果
            future.completeExceptionally(e);
        }
        // 2.2设置计算结果到future
        System.out.println("----" + Thread.currentThread().getName() + " set future result----");

    }, "thread-1").start();

    // 3.等待计算结果
    System.out.println(future.get());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

如上代码2.1.3表示当出现异常时把异常信息设置到future内部，这样代码3就会在抛出异常后终止。

其实我们还可以修改代码3为：

System.out.println(future.exceptionally(t -> "default").get());// 默认值
1

实现当出现异常时返回默认值。