CompletableFuture的基本用法

前言

• CompletableFuture是Java8新增的一个功能十分强大的工具类，它一方面实现了Future接口，另一方面也实现了CompletionStage接口，CompletionStage接口多达40中方法，为我们函数式编程
  流式调用提供支持。相较于FutureTask来做多任务更简洁了。

使用

完成了就通知我

• 核心代码

    /**
     * 完成了就通知我 ，手动
     *
     * @return
     */
    public String completeNotify() {
        CompletableFuture<Integer> future = new CompletableFuture<>();
        threadPoolTaskExecutor.execute(new AskThread(future));
        try {
            Integer result = future.get();
            System.out.println("result " + result);
            return result.toString();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    class AskThread implements Runnable {
        CompletableFuture<Integer> future;

        public AskThread(CompletableFuture<Integer> future) {
            this.future = future;
        }

        @Override
        public void run() {
            int res = 0;
            try {
                // 模拟长时间计算过程
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            res = 100;
            // 通知完成
            future.complete(res);

        }
    }

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

• 之前我利用这个功能完成了 高并发场景下请求合并(批量)
  的功能，可以参考下。

异步执行任务

• 核心代码

     public String asyncTask() {
        StopWatch stopWatch = new StopWatch("asyncTask");
        stopWatch.start("task");
        // 如果是runAsync 没有返回值
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> calc(50), threadPoolTaskExecutor);

        CompletableFuture<Integer> futureTwo = CompletableFuture.supplyAsync(() -> calc(60), threadPoolTaskExecutor);
        int result = 0;
        int res = 0;
        try {
            result = future.get();
            res = futureTwo.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }


        System.out.println(result + " " + res);
        stopWatch.stop();
        System.out.println(stopWatch.prettyPrint());
        System.out.println(stopWatch.getLastTaskTimeMillis());

        return result + " " + res;
    }
       public int calc(int param) {


        try {
            // 模拟耗时
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        if (EXCEPTION_PARAM == param){
            throw new RuntimeException("传了异常参数 "+param);
        }
        return param * 2;
    }

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

流式调用

    public String stream() {

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> calc(50), threadPoolTaskExecutor).
                thenApply((i) -> Integer.toString(i)).
                thenApply((str) -> "res " + str).
                thenAccept(System.out::println);
        try {
            future.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "done";
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

异常处理

     public String exception() {

        CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> calc(10))
                .exceptionally(ex -> {
                    System.out.println("异常信息 " + ex.toString());
                    return 0;
                })
                .thenApply((i) -> Integer.toString(i)).
                thenApply((str) -> "res " + str).
                thenAccept(System.out::println);
        try {
            future.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }


        return "done";
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

组合多个CompletableFuture

   public String compose(){

        CompletableFuture future = CompletableFuture.supplyAsync(()->calc(50),threadPoolTaskExecutor)
                .thenCompose((i)->CompletableFuture.supplyAsync(()->calc(i),threadPoolTaskExecutor))
                .thenApply((str)->"res " + str)
                .thenAccept(System.out::println);
        try {
            future.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        return "done";
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

小结

• CompletableFuture很强大，如果写异步任务相比FutureTask更简洁。
• 源码地址：https://github.com/1030907690/CompletableFuture-Sample
• 视频地址：https://www.bilibili.com/video/BV1B24y1C7bT/