Java异步编排CompletableFuture简述

CompletableFuture

介绍

在Java8之后提供了一个类以帮助我们进行多线程的异步编排工作，这就是CompletableFuture类，该类优化了我们对于异步任务的掌控，使我们可以自由编排任务先后顺序，返回值，中间条件

继承和实现

该类实现了：

1. Future
2. CompletionStage

主要方法

QuickStart

public class Test {

    static ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5,
            10,
            10,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        System.out.println(Thread.currentThread().getName());
        //无返回结果
        CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
        }, executor).whenComplete((res, exec) -> {
            //res为返回结构，exec为异常信息
            System.out.println(Thread.currentThread().getName() + "!");
        });
        //有返回结果，返回结果需要阻塞等待
        final CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            return "test supply";
        }, executor).whenCompleteAsync((res, exec) -> {
            System.out.println("r:" + res);
            System.out.println("e:" + exec);
            System.out.println(Thread.currentThread().getName() + "??");
        }, executor).exceptionally((e)->{
            System.out.println("e = " + e);
            return "ok";
        });
        System.out.println(supplyAsync.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（双线程）

public class Test2 {
    static ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5,
            10,
            10,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );
    public static void main(String[] args) {
        final CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            System.out.println("task1" + Thread.currentThread().getName());

        }, executor).thenRunAsync(() -> {
            System.out.println("task2" + Thread.currentThread().getName());
        }).thenAccept((e) -> {
            System.out.println(e);
            System.out.println("task3" + Thread.currentThread().getName());
        });

        final CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("task4" + Thread.currentThread().getName());
            return "task4";
        }, executor).thenApplyAsync(res -> {
            System.out.println(res);
            return "task5";
        }).thenAccept(res -> {
            System.out.println(res);
        });

        future1.runAfterBothAsync(future2,()->{
            System.out.println("task6");
        },executor);
    }
}

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

练习（任意任务线完成执行新任务）

public class Test3 {
    static ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5,
            10,
            10,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            Executors.defaultThreadFactory(),
            new ThreadPoolExecutor.AbortPolicy()
    );
    public static void main(String[] args) {
        final CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            System.out.println("task1" + Thread.currentThread().getName());

        }, executor).thenRunAsync(() -> {
            System.out.println("task2" + Thread.currentThread().getName());
        }).thenAccept((e) -> {
            System.out.println(e);
            System.out.println("task3" + Thread.currentThread().getName());
        });

        final CompletableFuture<Void> future2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("task4" + Thread.currentThread().getName());
            return "task4";
        }, executor).thenApplyAsync(res -> {
            System.out.println(res);
            return "task5";
        }).thenAccept(res -> {
            System.out.println(res);
        });

//        future1.runAfterEither(future2,()->{
//            System.out.println(Thread.currentThread().getName());
//            System.out.println("either task");
//        });
        future1.runAfterEitherAsync(future2,()->{
            System.out.println(Thread.currentThread().getName());
            System.out.println("either task");
        });
    }
}
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

多任务编排

前面我们了解到双任务编排，那么大于等于3条任务线的时候我们就需要多任务编排了，主要用到anyOf和allOf方法进行，最后完成后会回到主线程

final CompletableFuture<Void> newF = CompletableFuture.allOf(future1, future2, future3);
//用于对任务线进行阻塞,join和get都可以，看场景，单纯阻塞无返回值建议join
newF.join();
newF.get();
1
2
3
4