CompletableFuture 使用（一）

发表于

如果我们有以下需求：

  1. 从数据库的User表获取所有用户的id
  2. 根据用户的id查询所有用户相关联的信息
  3. 获取信息后，处理信息
  4. 等待所有用户的信息处理完成后，完成下一步动作
  5. 在处理过程中主线程无需等待整个信息的处理完成

通过Future来处理：

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
scss复制代码public class CompletableFutureTest {
    public static final String USER_MSG_FORMAT = "用户信息%d";
    public static final String USER_MSG_START_FORMAT = "正在获取用户信息%d的信息";
    public static final String USER_MSG_END_FORMAT = "获取结束";

    private static ExecutorService executor = Executors.newFixedThreadPool(10);

    public static void main(String[] args) throws Exception {
        Future<List<Integer>> userIdListFuture = executor.submit(CompletableFutureTest::getUserIdList);
        mainThreadDo();
        List<Integer> userIdList = userIdListFuture.get();
        List<Future<String>> userMsg = new ArrayList<>();
        userIdList.forEach(userId -> userMsg.add(executor.submit(() -> getUserMsg(userId))));
        mainThreadDo();
        for (Future<String> res : userMsg) {
            System.out.println(Thread.currentThread().getName() + " " + res.get());
        }
        mainThreadDo();
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.DAYS);
        System.out.println(Thread.currentThread().getName() + " 业务处理完毕");
    }

    public static void mainThreadDo() {
        System.out.println(Thread.currentThread().getName() + " 主线程开始执行别的业务逻辑");
        sleep();
        System.out.println(Thread.currentThread().getName() + " 主线程结束执行别的业务逻辑");
    }

    public static String getUserMsg(Integer userId) {
        System.out.println(Thread.currentThread().getName() + " " + String.format(USER_MSG_START_FORMAT, userId));
        sleep();
        System.out.println(Thread.currentThread().getName() + " " + USER_MSG_END_FORMAT);
        return String.format(USER_MSG_FORMAT, userId);
    }

    public static List<Integer> getUserIdList() {
        return Lists.newArrayList(1, 2, 3, 4, 5);
    }

    //模拟数据库延时操作
    public static void sleep() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

打印结果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
css复制代码main 主线程开始执行别的业务逻辑
main 主线程结束执行别的业务逻辑
main 主线程开始执行别的业务逻辑
pool-1-thread-4 正在获取用户信息3的信息
pool-1-thread-3 正在获取用户信息2的信息
pool-1-thread-2 正在获取用户信息1的信息
pool-1-thread-5 正在获取用户信息4的信息
pool-1-thread-6 正在获取用户信息5的信息
main 主线程结束执行别的业务逻辑
pool-1-thread-4 获取结束
pool-1-thread-5 获取结束
pool-1-thread-6 获取结束
pool-1-thread-3 获取结束
pool-1-thread-2 获取结束
main 用户信息1
main 用户信息2
main 用户信息3
main 用户信息4
main 用户信息5
main 主线程开始执行别的业务逻辑
main 主线程结束执行别的业务逻辑
main 业务处理完毕

image.png

通过 CompletableFuture 来处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
scss复制代码public static void main(String[] args) throws Exception {
    System.out.println(Thread.currentThread().getName() + " 业务开始完毕");
    List<CompletableFuture<Void>> completableFutureList = CompletableFuture.supplyAsync(CompletableFutureTest2::getUserIdList)//将getUserIdList封装成Callable放入ForkJoinPool中执行
            .get()//等待执行结果
            .stream()
            .map(userId -> CompletableFuture.supplyAsync(() -> getUserMsg(userId)).thenAccept(System.out::println))//将stream List<UserId> 转换为List<CompletableFuture>
            .collect(Collectors.toList());
    mainThreadDo();
    //将所有的List<CompletableFuture>包装成一个 CompletableFuture
    CompletableFuture<Void> completableFuture = CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[0]));
    completableFuture.thenRun(() -> mainThreadDo());//将封装的一个CompletableFuture安装一个执行结束后的回调动作mainThreadDo()
    completableFuture.get();//等待completableFuture执行完成
    System.out.println(Thread.currentThread().getName() + " 业务处理完毕");
}

CompletableFuture执行回调示例图
image.png

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%